From 9ef1ba3702e62d315a4382a6dd606b0ef4ba03dc Mon Sep 17 00:00:00 2001 From: dtoro Date: Sat, 11 Jul 2026 18:27:29 +0200 Subject: [PATCH] fix(concurrency): scope assent/destructive windows to session, not just agent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix 1 of plans/2026-07-11-concurrent-task-execution.md — the safety-critical one. The assent window (and destructive window) were keyed purely by agent id ("assent_window.agent:"). With one agent:nomos entity serving every concurrent task, this meant approving Task A's plan opened a window that ANY concurrently-running task's config-mutation/destructive actions could also ride, auto-executing without their own approval. - store.go / agent.go: assentWindowActive/openAssentWindow and destructiveWindowActive/openDestructiveWindow/destructiveWindowKey all gain a sessionID parameter; keys become "assent_window.agent:.session:" and "destructive_window.agent:.target:.session:". Missing session id fails closed (no window) rather than falling back to the old agent-wide key. - continue.go: the auto-continuation worker's window check moved from once- per-batch to once-per-pending-item, scoped to that item's own session — it was previously checking ONE agent-wide window for a batch that can span multiple tasks. - agent.go tool-dispatch: injects `_session_id` into a COPY of the wire args sent to the MCP server (never into the args used for the emitted/logged/ persisted tool call, and never part of any tool's declared InputSchema — invisible to the model) so the gating checks on the OTHER side of the process boundary know which task is asking. - internal/mcp/server.go: assentWindowActive/destructiveWindowActive/ classifyAndGate gain the same sessionID parameter, read from args["_session_id"] at the three call sites (request_execution's apt_upgrade/pct_create branches, and the shared classifyAndGate used by restart/pct_exec/systemctl/run). Verified against the live stack with the exact scenario from the plan: opened an assent window for session A only, then called `run` with an identical config-mutation command for session A (window open) and session B (same agent, no window). A auto-ran (execution status completed); B correctly queued for approval (pending_approval) instead of bleeding through — proven at both the MCP response text and the executions table. Co-Authored-By: Claude Opus 4.8 --- cmd/nomos/agent.go | 35 +++++++++++++++------- cmd/nomos/continue.go | 14 +++++---- cmd/nomos/store.go | 67 ++++++++++++++++++++++++++---------------- internal/mcp/server.go | 58 +++++++++++++++++++++--------------- 4 files changed, 108 insertions(+), 66 deletions(-) diff --git a/cmd/nomos/agent.go b/cmd/nomos/agent.go index 886cdf5..56f0c1c 100644 --- a/cmd/nomos/agent.go +++ b/cmd/nomos/agent.go @@ -125,12 +125,15 @@ const assentWindowDuration = 30 * time.Minute // openAssentWindow records an active assent window in autonomy_settings so // the MCP run tool (separate process) can check it before requiring approval -// for config_mutation commands. Key is scoped to this agent's UUID. -func (a *agent) openAssentWindow(ctx context.Context) { - if a.store == nil || a.store.pool == nil || a.agentID == uuid.Nil { +// for config_mutation commands. Key is scoped to this agent's UUID AND this +// session/task — see store.go's assentWindowActive for why: without the +// session dimension, approving one task's plan would silently auto-run +// unapproved actions in any other concurrently-running task. +func (a *agent) openAssentWindow(ctx context.Context, sessionID string) { + if a.store == nil || a.store.pool == nil || a.agentID == uuid.Nil || sessionID == "" { return } - key := "assent_window.agent:" + a.agentID.String() + key := assentWindowKey(a.agentID, sessionID) expires := time.Now().Add(assentWindowDuration).UTC().Format(time.RFC3339) _, err := a.store.pool.Exec(ctx, `INSERT INTO autonomy_settings (key, value) VALUES ($1, $2) @@ -138,7 +141,7 @@ func (a *agent) openAssentWindow(ctx context.Context) { if err != nil { slog.Warn("nomos: openAssentWindow", "error", err) } else { - slog.Info("nomos: assent window opened", "agent", a.agentID, "expires", expires) + slog.Info("nomos: assent window opened", "agent", a.agentID, "session", sessionID, "expires", expires) } } @@ -242,15 +245,15 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s if p.destructive && typedConfirm { if execUUID, perr := uuid.Parse(p.execID); perr == nil { if target := a.store.executionTarget(ctx, execUUID); target != "" { - a.store.openDestructiveWindow(ctx, a.agentID, target) - slog.Info("nomos: destructive window opened", "agent", a.agentID, "target", target) + a.store.openDestructiveWindow(ctx, a.agentID, target, sessionID) + slog.Info("nomos: destructive window opened", "agent", a.agentID, "target", target, "session", sessionID) } } } } } if len(granted) > 0 { - a.openAssentWindow(ctx) + a.openAssentWindow(ctx, sessionID) note := fmt.Sprintf("[System: the operator's message approved pending execution(s) %s via chat assent — they are now running. An assent window is now active for 30 minutes: config_mutation commands will auto-run without re-approval. Do not re-request or call request_execution/run again for these; check get_execution_status if you need the outcome. CONTINUE executing the full plan — do not stop and wait for 'continue' after each step. Only surface to the operator for destructive actions (need typed confirmation) or if you're genuinely stuck after trying alternatives.]", strings.Join(granted, ", ")) messages = append(messages, openai.SystemMessage(note)) } @@ -266,7 +269,7 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s // the operator approved — go execute the plan now. note := "[System: The operator approved your proposed plan. Execute it now — call request_execution or run to carry out the steps you described. Do not re-describe the plan or ask for confirmation again. The assent window is active: config_mutation commands will auto-run once you create them.]" messages = append(messages, openai.SystemMessage(note)) - a.openAssentWindow(ctx) + a.openAssentWindow(ctx, sessionID) } // Worker continuation: append the finished-execution note so the model @@ -367,7 +370,19 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s if localRes, handled := a.handleTaskTool(ctx, sessionID, tc.Function.Name, args); handled { result = localRes } else { - result, callErr = a.client.callTool(tc.Function.Name, args) + // _session_id rides along on the wire call only — never in + // `args` (which is what gets emitted/logged/persisted as the + // model's own tool call) — so the MCP-side assent/destructive + // window checks can scope to THIS task instead of bleeding + // across every concurrently-running one sharing this agent + // identity. Not part of any tool's declared InputSchema, so + // the model never sees or supplies it. + wireArgs := make(map[string]any, len(args)+1) + for k, v := range args { + wireArgs[k] = v + } + wireArgs["_session_id"] = sessionID + result, callErr = a.client.callTool(tc.Function.Name, wireArgs) } elapsed := int(time.Since(start).Milliseconds()) diff --git a/cmd/nomos/continue.go b/cmd/nomos/continue.go index 0176942..5113667 100644 --- a/cmd/nomos/continue.go +++ b/cmd/nomos/continue.go @@ -57,13 +57,15 @@ func (a *agent) runContinuationWorker(ctx context.Context) { func (a *agent) processContinuations(ctx context.Context) { pending := a.store.pendingContinuations(ctx, 5) - windowOpen := a.store.assentWindowActive(ctx, a.agentID) for _, p := range pending { - // Scope gate: only auto-continue while an approved plan is active. - // A finished one-off execution with no window is left as-is (marked - // continued so we don't re-check it forever) — the operator decides - // what happens next, as today. - if !windowOpen { + // Scope gate: only auto-continue while an approved plan is active FOR + // THIS SESSION. Checked per-item, not once for the whole batch — with + // multiple tasks in flight, one task's open window must never cover a + // pending continuation belonging to a different task. + if !a.store.assentWindowActive(ctx, a.agentID, p.SessionID) { + // A finished one-off execution with no window is left as-is + // (marked continued so we don't re-check it forever) — the + // operator decides what happens next, as today. a.store.markContinued(ctx, p.ExecID) continue } diff --git a/cmd/nomos/store.go b/cmd/nomos/store.go index 7c1bbd6..691a38f 100644 --- a/cmd/nomos/store.go +++ b/cmd/nomos/store.go @@ -810,59 +810,74 @@ func (s *store) markContinued(ctx context.Context, execID uuid.UUID) { s.pool.Exec(ctx, `UPDATE nomos_plan_executions SET continued_at = now() WHERE execution_id = $1`, execID) } -// assentWindowActive reports whether this agent currently has an open assent -// window — the scope gate for auto-continuation. We only auto-continue -// executions that are part of an approved plan, never stray one-off actions. -func (s *store) assentWindowActive(ctx context.Context, agentID uuid.UUID) bool { - if s == nil || agentID == uuid.Nil { - return false +// assentWindowActive reports whether THIS TASK currently has an open assent +// window — the scope gate for auto-continuation. Scoped by session, not just +// agent: with a single agent:nomos entity serving every concurrent task, an +// agent-only key would let approving Task A's plan silently auto-run +// unapproved config-mutation actions in a concurrently-running Task B. We +// only auto-continue executions that are part of THIS session's approved +// plan, never a stray action from another task riding the same window. +func (s *store) assentWindowActive(ctx context.Context, agentID uuid.UUID, sessionID string) bool { + if s == nil || agentID == uuid.Nil || sessionID == "" { + return false // fail closed: no session to scope to means no window } var expires time.Time - key := "assent_window.agent:" + agentID.String() + key := assentWindowKey(agentID, sessionID) if err := s.pool.QueryRow(ctx, `SELECT value::timestamptz FROM autonomy_settings WHERE key = $1`, key).Scan(&expires); err != nil { return false } return time.Now().Before(expires) } +// assentWindowKey scopes the grant to one agent AND one session/task — see +// assentWindowActive. Must match internal/mcp/server.go's copy (mirrored +// there, not shared, since the two are separate Go packages/binaries reading +// the same autonomy_settings row). +func assentWindowKey(agentID uuid.UUID, sessionID string) string { + return "assent_window.agent:" + agentID.String() + ".session:" + sessionID +} + // destructiveWindowDuration is intentionally shorter than the general assent // window (30 min): it's a narrow, scoped grant for a multi-step DESTRUCTIVE // recovery (e.g. "stop then destroy this specific half-provisioned // container"), not a standing license to destroy things. const destructiveWindowDuration = 15 * time.Minute -// destructiveWindowKey scopes the grant to one agent AND one target entity — -// an explicit typed confirmation ("I confirm") for a destructive action on -// target X must never be read as authorizing a destructive action on target Y. -func destructiveWindowKey(agentID uuid.UUID, targetSlug string) string { - return "destructive_window.agent:" + agentID.String() + ".target:" + targetSlug +// destructiveWindowKey scopes the grant to one agent, one target entity, AND +// one session/task — an explicit typed confirmation ("I confirm") for a +// destructive action on target X in task A must never be read as authorizing +// a destructive action on target X from a DIFFERENT concurrently-running +// task B, even though both share the same agent identity. +func destructiveWindowKey(agentID uuid.UUID, targetSlug, sessionID string) string { + return "destructive_window.agent:" + agentID.String() + ".target:" + targetSlug + ".session:" + sessionID } -// openDestructiveWindow records a short, target-scoped grant after an -// operator's EXPLICIT typed confirmation (never loose assent) authorized a -// destructive action. Real case this exists for: recovering a failed destroy -// took "stop" (destructive) then "destroy" (destructive) — same container, -// two separate typed-confirmation round trips, because each was gated -// independently. One explicit confirmation on a target should cover the -// short follow-up sequence needed to finish what was just confirmed. -func (s *store) openDestructiveWindow(ctx context.Context, agentID uuid.UUID, targetSlug string) { - if s == nil || agentID == uuid.Nil || targetSlug == "" { +// openDestructiveWindow records a short, target-and-session-scoped grant +// after an operator's EXPLICIT typed confirmation (never loose assent) +// authorized a destructive action. Real case this exists for: recovering a +// failed destroy took "stop" (destructive) then "destroy" (destructive) — +// same container, two separate typed-confirmation round trips, because each +// was gated independently. One explicit confirmation on a target should +// cover the short follow-up sequence needed to finish what was just +// confirmed — but only within the task that got the confirmation. +func (s *store) openDestructiveWindow(ctx context.Context, agentID uuid.UUID, targetSlug, sessionID string) { + if s == nil || agentID == uuid.Nil || targetSlug == "" || sessionID == "" { return } expires := time.Now().Add(destructiveWindowDuration).UTC().Format(time.RFC3339) s.pool.Exec(ctx, `INSERT INTO autonomy_settings (key, value) VALUES ($1, $2) - ON CONFLICT (key) DO UPDATE SET value = $2`, destructiveWindowKey(agentID, targetSlug), expires) + ON CONFLICT (key) DO UPDATE SET value = $2`, destructiveWindowKey(agentID, targetSlug, sessionID), expires) } // destructiveWindowActive reports whether target has a live, explicitly- -// confirmed destructive grant for this agent. -func (s *store) destructiveWindowActive(ctx context.Context, agentID uuid.UUID, targetSlug string) bool { - if s == nil || agentID == uuid.Nil || targetSlug == "" { +// confirmed destructive grant for this agent within this session/task. +func (s *store) destructiveWindowActive(ctx context.Context, agentID uuid.UUID, targetSlug, sessionID string) bool { + if s == nil || agentID == uuid.Nil || targetSlug == "" || sessionID == "" { return false } var expires time.Time if err := s.pool.QueryRow(ctx, `SELECT value::timestamptz FROM autonomy_settings WHERE key = $1`, - destructiveWindowKey(agentID, targetSlug)).Scan(&expires); err != nil { + destructiveWindowKey(agentID, targetSlug, sessionID)).Scan(&expires); err != nil { return false } return time.Now().Before(expires) diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 6976a00..e3bd92f 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -356,6 +356,7 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server { targetSlug, _ := args["target"].(string) action, _ := args["action"].(string) params, _ := args["params"].(string) + sessionID, _ := args["_session_id"].(string) if targetSlug == "" || action == "" { return textResult("error: target and action required"), nil } @@ -388,7 +389,7 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server { cmd = fmt.Sprintf("systemctl %s %s; sleep 1; systemctl is-active %s", params, svc, svc) purpose = "systemctl " + params + " " + svc } - return classifyAndGate(ctx, pool, agentID, targetID, targetSlug, cmd, purpose, ""), nil + return classifyAndGate(ctx, pool, agentID, targetID, targetSlug, cmd, purpose, "", sessionID), nil } // Deduplicate: if a pending execution already exists for the same @@ -451,7 +452,7 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server { return textResult("apt audit:\n" + out), nil } // During an active assent window, auto-approve. - if assentWindowActive(ctx, pool, agentID) { + if assentWindowActive(ctx, pool, agentID, sessionID) { pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id) createApproval(ctx, pool, id, targetID, "apt_upgrade", params, "config_mutation") // Do NOT pre-flip approvals/executions status here (that was @@ -485,7 +486,7 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server { case "pct_create": // During an active assent window, auto-approve and execute // instead of queuing — the operator already approved the plan. - if assentWindowActive(ctx, pool, agentID) { + if assentWindowActive(ctx, pool, agentID, sessionID) { pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id) createApproval(ctx, pool, id, targetID, "pct_create", params, "config_mutation") // See the apt_upgrade case above for why there's no @@ -517,6 +518,7 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server { command, _ := args["command"].(string) purpose, _ := args["purpose"].(string) declaredRisk, _ := args["declared_risk"].(string) + sessionID, _ := args["_session_id"].(string) if targetSlug == "" || command == "" { return textResult("error: target and command are required"), nil } @@ -526,7 +528,7 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server { return textResult(fmt.Sprintf("target not found: %s", targetSlug)), nil } - return classifyAndGate(ctx, pool, agentID, targetID, targetSlug, command, purpose, declaredRisk), nil + return classifyAndGate(ctx, pool, agentID, targetID, targetSlug, command, purpose, declaredRisk, sessionID), nil }) register(&mcp.Tool{Name: "http_get", Description: "Fetch a public web page or raw file (e.g. a GitHub README/raw URL) and return sanitized text. Use this to research how to deploy a service before provisioning. HTTP/HTTPS only; body is truncated to ~16KB.", @@ -1308,7 +1310,7 @@ func resolveExecTarget(ctx context.Context, pool *db.Pool, targetSlug string) (h // fleet's reverse proxy) executed instantly with no approval at all. Routing // every mutating path through the same classifier + approval-queue logic // closes that gap without special-casing each caller. -func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.UUID, targetSlug, command, purpose, declaredRisk string) *mcp.CallToolResult { +func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.UUID, targetSlug, command, purpose, declaredRisk, sessionID string) *mcp.CallToolResult { riskClass := policy.ClassifyCommand(command, declaredRisk) runParams, _ := json.Marshal(map[string]string{"command": command, "purpose": purpose}) actionCol := "run:" + string(runParams) @@ -1360,7 +1362,7 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid. // operator approved the overall direction; individual config steps // within the window don't each need a separate yes. Destructive // commands never auto-run, regardless of window. - if riskClass == policy.RiskConfigMutation && assentWindowActive(ctx, pool, agentID) { + if riskClass == policy.RiskConfigMutation && assentWindowActive(ctx, pool, agentID, sessionID) { host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug) if rerr != nil { pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error())) @@ -1382,7 +1384,7 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid. // recovery (e.g. a failed destroy needing stop, then destroy) so the // operator isn't asked to re-type "I confirm" for every single command // against the thing they just confirmed. - if riskClass == policy.RiskDestructive && destructiveWindowActive(ctx, pool, agentID, targetSlug) { + if riskClass == policy.RiskDestructive && destructiveWindowActive(ctx, pool, agentID, targetSlug, sessionID) { host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug) if rerr != nil { pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error())) @@ -1450,18 +1452,25 @@ func executeApprovedViaAPI(ctx context.Context, execID uuid.UUID, targetSlug, ac } // assentWindowActive checks whether the operator has recently approved a plan -// in this agent's chat session. The agent sets an assent_window.agent: -// key in autonomy_settings with an expiry timestamp when chat-assent grants -// a pending execution. While active, config_mutation commands auto-run -// without re-approval — the operator approved the overall plan, not each step. -func assentWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID) bool { - if agentID == uuid.Nil { - return false +// in THIS TASK's chat session. The agent sets an +// assent_window.agent:.session: key in autonomy_settings with an +// expiry timestamp when chat-assent grants a pending execution. While +// active, config_mutation commands auto-run without re-approval — the +// operator approved the overall plan, not each step. Scoped by session, not +// just agent: with one agent:nomos entity serving every concurrent task, an +// agent-only key would let approving Task A's plan silently auto-run +// unapproved actions from a concurrently-running Task B. sessionID comes +// from the `_session_id` nomos injects into every tool call's wire args +// (never part of any tool's declared InputSchema, so the model never +// supplies or sees it) — see cmd/nomos/agent.go's tool dispatch loop. +func assentWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID, sessionID string) bool { + if agentID == uuid.Nil || sessionID == "" { + return false // fail closed: no session to scope to means no window } var expiresStr string err := pool.QueryRow(ctx, "SELECT value FROM autonomy_settings WHERE key = $1", - "assent_window.agent:"+agentID.String()).Scan(&expiresStr) + "assent_window.agent:"+agentID.String()+".session:"+sessionID).Scan(&expiresStr) if err != nil { return false } @@ -1473,20 +1482,21 @@ func assentWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID) b } // destructiveWindowActive reports whether targetSlug has a live, explicitly- -// confirmed destructive grant for this agent. Key format -// ("destructive_window.agent:.target:") must match -// cmd/nomos/store.go's openDestructiveWindow — both processes read/write the -// same autonomy_settings row. Scoped to one target so a typed confirmation -// for destroying container A can never be read as authorizing anything -// against container B. -func destructiveWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID, targetSlug string) bool { - if agentID == uuid.Nil || targetSlug == "" { +// confirmed destructive grant for this agent WITHIN THIS SESSION/TASK. Key +// format ("destructive_window.agent:.target:.session:") must +// match cmd/nomos/store.go's openDestructiveWindow — both processes +// read/write the same autonomy_settings row. Scoped to one target AND one +// session so a typed confirmation for destroying container A in task X can +// never be read as authorizing anything against container A from a +// different, concurrently-running task Y. +func destructiveWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID, targetSlug, sessionID string) bool { + if agentID == uuid.Nil || targetSlug == "" || sessionID == "" { return false } var expiresStr string err := pool.QueryRow(ctx, "SELECT value FROM autonomy_settings WHERE key = $1", - "destructive_window.agent:"+agentID.String()+".target:"+targetSlug).Scan(&expiresStr) + "destructive_window.agent:"+agentID.String()+".target:"+targetSlug+".session:"+sessionID).Scan(&expiresStr) if err != nil { return false }