dtoro/oikos
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix(concurrency): scope assent/destructive windows to session, not just agent

Browse Source 
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:<uuid>"). 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:<id>.session:<sessionID>" and
  "destructive_window.agent:<id>.target:<slug>.session:<sessionID>". 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 <noreply@anthropic.com>
This commit is contained in:
authentik Default Admin
2026-07-11 18:27:29 +02:00
parent 6932eb5eed
commit 9ef1ba3702
4 changed files with 108 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
cmd/nomos/agent.go
View File

@@ -125,12 +125,15 @@ const assentWindowDuration = 30 * time.Minute
// openAssentWindow records an active assent window in autonomy_settings so // openAssentWindow records an active assent window in autonomy_settings so
// the MCP run tool (separate process) can check it before requiring approval // the MCP run tool (separate process) can check it before requiring approval
// for config_mutation commands. Key is scoped to this agent's UUID. // for config_mutation commands. Key is scoped to this agent's UUID AND this
func (a *agent) openAssentWindow(ctx context.Context) { // session/task — see store.go's assentWindowActive for why: without the
if a.store == nil || a.store.pool == nil || a.agentID == uuid.Nil { // 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 return
} }
key := "assent_window.agent:" + a.agentID.String() key := assentWindowKey(a.agentID, sessionID)
expires := time.Now().Add(assentWindowDuration).UTC().Format(time.RFC3339) expires := time.Now().Add(assentWindowDuration).UTC().Format(time.RFC3339)
_, err := a.store.pool.Exec(ctx, _, err := a.store.pool.Exec(ctx,
`INSERT INTO autonomy_settings (key, value) VALUES ($1, $2) `INSERT INTO autonomy_settings (key, value) VALUES ($1, $2)
@@ -138,7 +141,7 @@ func (a *agent) openAssentWindow(ctx context.Context) {
if err != nil { if err != nil {
slog.Warn("nomos: openAssentWindow", "error", err) slog.Warn("nomos: openAssentWindow", "error", err)
} else { } 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 p.destructive && typedConfirm {
if execUUID, perr := uuid.Parse(p.execID); perr == nil { if execUUID, perr := uuid.Parse(p.execID); perr == nil {
if target := a.store.executionTarget(ctx, execUUID); target != "" { if target := a.store.executionTarget(ctx, execUUID); target != "" {
a.store.openDestructiveWindow(ctx, a.agentID, target) a.store.openDestructiveWindow(ctx, a.agentID, target, sessionID)
slog.Info("nomos: destructive window opened", "agent", a.agentID, "target", target) slog.Info("nomos: destructive window opened", "agent", a.agentID, "target", target, "session", sessionID)
} }
} }
} }
} }
} }
if len(granted) > 0 { 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, ", ")) 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)) 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. // 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.]" 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)) messages = append(messages, openai.SystemMessage(note))
a.openAssentWindow(ctx) a.openAssentWindow(ctx, sessionID)
} }
// Worker continuation: append the finished-execution note so the model // 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 { if localRes, handled := a.handleTaskTool(ctx, sessionID, tc.Function.Name, args); handled {
result = localRes result = localRes
} else { } 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()) elapsed := int(time.Since(start).Milliseconds())

14
cmd/nomos/continue.go
View File

@@ -57,13 +57,15 @@ func (a *agent) runContinuationWorker(ctx context.Context) {
func (a *agent) processContinuations(ctx context.Context) { func (a *agent) processContinuations(ctx context.Context) {
pending := a.store.pendingContinuations(ctx, 5) pending := a.store.pendingContinuations(ctx, 5)
windowOpen := a.store.assentWindowActive(ctx, a.agentID)
for _, p := range pending { for _, p := range pending {
// Scope gate: only auto-continue while an approved plan is active. // Scope gate: only auto-continue while an approved plan is active FOR
// A finished one-off execution with no window is left as-is (marked // THIS SESSION. Checked per-item, not once for the whole batch — with
// continued so we don't re-check it forever) — the operator decides // multiple tasks in flight, one task's open window must never cover a
// what happens next, as today. // pending continuation belonging to a different task.
if !windowOpen { 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) a.store.markContinued(ctx, p.ExecID)
continue continue
} }

67
cmd/nomos/store.go
View File

@@ -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) 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 // assentWindowActive reports whether THIS TASK currently has an open assent
// window — the scope gate for auto-continuation. We only auto-continue // window — the scope gate for auto-continuation. Scoped by session, not just
// executions that are part of an approved plan, never stray one-off actions. // agent: with a single agent:nomos entity serving every concurrent task, an
func (s *store) assentWindowActive(ctx context.Context, agentID uuid.UUID) bool { // agent-only key would let approving Task A's plan silently auto-run
if s == nil || agentID == uuid.Nil { // unapproved config-mutation actions in a concurrently-running Task B. We
return false // 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 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 { if err := s.pool.QueryRow(ctx, `SELECT value::timestamptz FROM autonomy_settings WHERE key = $1`, key).Scan(&expires); err != nil {
return false return false
} }
return time.Now().Before(expires) 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 // destructiveWindowDuration is intentionally shorter than the general assent
// window (30 min): it's a narrow, scoped grant for a multi-step DESTRUCTIVE // window (30 min): it's a narrow, scoped grant for a multi-step DESTRUCTIVE
// recovery (e.g. "stop then destroy this specific half-provisioned // recovery (e.g. "stop then destroy this specific half-provisioned
// container"), not a standing license to destroy things. // container"), not a standing license to destroy things.
const destructiveWindowDuration = 15 * time.Minute const destructiveWindowDuration = 15 * time.Minute
// destructiveWindowKey scopes the grant to one agent AND one target entity — // destructiveWindowKey scopes the grant to one agent, one target entity, AND
// an explicit typed confirmation ("I confirm") for a destructive action on // one session/task — an explicit typed confirmation ("I confirm") for a
// target X must never be read as authorizing a destructive action on target Y. // destructive action on target X in task A must never be read as authorizing
func destructiveWindowKey(agentID uuid.UUID, targetSlug string) string { // a destructive action on target X from a DIFFERENT concurrently-running
return "destructive_window.agent:" + agentID.String() + ".target:" + targetSlug // 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 // openDestructiveWindow records a short, target-and-session-scoped grant
// operator's EXPLICIT typed confirmation (never loose assent) authorized a // after an operator's EXPLICIT typed confirmation (never loose assent)
// destructive action. Real case this exists for: recovering a failed destroy // authorized a destructive action. Real case this exists for: recovering a
// took "stop" (destructive) then "destroy" (destructive) — same container, // failed destroy took "stop" (destructive) then "destroy" (destructive) —
// two separate typed-confirmation round trips, because each was gated // same container, two separate typed-confirmation round trips, because each
// independently. One explicit confirmation on a target should cover the // was gated independently. One explicit confirmation on a target should
// short follow-up sequence needed to finish what was just confirmed. // cover the short follow-up sequence needed to finish what was just
func (s *store) openDestructiveWindow(ctx context.Context, agentID uuid.UUID, targetSlug string) { // confirmed — but only within the task that got the confirmation.
if s == nil || agentID == uuid.Nil || targetSlug == "" { func (s *store) openDestructiveWindow(ctx context.Context, agentID uuid.UUID, targetSlug, sessionID string) {
if s == nil || agentID == uuid.Nil || targetSlug == "" || sessionID == "" {
return return
} }
expires := time.Now().Add(destructiveWindowDuration).UTC().Format(time.RFC3339) expires := time.Now().Add(destructiveWindowDuration).UTC().Format(time.RFC3339)
s.pool.Exec(ctx, `INSERT INTO autonomy_settings (key, value) VALUES ($1, $2) 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- // destructiveWindowActive reports whether target has a live, explicitly-
// confirmed destructive grant for this agent. // confirmed destructive grant for this agent within this session/task.
func (s *store) destructiveWindowActive(ctx context.Context, agentID uuid.UUID, targetSlug string) bool { func (s *store) destructiveWindowActive(ctx context.Context, agentID uuid.UUID, targetSlug, sessionID string) bool {
if s == nil || agentID == uuid.Nil || targetSlug == "" { if s == nil || agentID == uuid.Nil || targetSlug == "" || sessionID == "" {
return false return false
} }
var expires time.Time var expires time.Time
if err := s.pool.QueryRow(ctx, `SELECT value::timestamptz FROM autonomy_settings WHERE key = $1`, 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 false
} }
return time.Now().Before(expires) return time.Now().Before(expires)

58
internal/mcp/server.go
View File

@@ -356,6 +356,7 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server {
targetSlug, _ := args["target"].(string) targetSlug, _ := args["target"].(string)
action, _ := args["action"].(string) action, _ := args["action"].(string)
params, _ := args["params"].(string) params, _ := args["params"].(string)
sessionID, _ := args["_session_id"].(string)
if targetSlug == "" || action == "" { if targetSlug == "" || action == "" {
return textResult("error: target and action required"), nil 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) cmd = fmt.Sprintf("systemctl %s %s; sleep 1; systemctl is-active %s", params, svc, svc)
purpose = "systemctl " + params + " " + 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 // 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 return textResult("apt audit:\n" + out), nil
} }
// During an active assent window, auto-approve. // 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) 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") createApproval(ctx, pool, id, targetID, "apt_upgrade", params, "config_mutation")
// Do NOT pre-flip approvals/executions status here (that was // 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": case "pct_create":
// During an active assent window, auto-approve and execute // During an active assent window, auto-approve and execute
// instead of queuing — the operator already approved the plan. // 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) 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") createApproval(ctx, pool, id, targetID, "pct_create", params, "config_mutation")
// See the apt_upgrade case above for why there's no // 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) command, _ := args["command"].(string)
purpose, _ := args["purpose"].(string) purpose, _ := args["purpose"].(string)
declaredRisk, _ := args["declared_risk"].(string) declaredRisk, _ := args["declared_risk"].(string)
sessionID, _ := args["_session_id"].(string)
if targetSlug == "" || command == "" { if targetSlug == "" || command == "" {
return textResult("error: target and command are required"), nil 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 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.", 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 // fleet's reverse proxy) executed instantly with no approval at all. Routing
// every mutating path through the same classifier + approval-queue logic // every mutating path through the same classifier + approval-queue logic
// closes that gap without special-casing each caller. // 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) riskClass := policy.ClassifyCommand(command, declaredRisk)
runParams, _ := json.Marshal(map[string]string{"command": command, "purpose": purpose}) runParams, _ := json.Marshal(map[string]string{"command": command, "purpose": purpose})
actionCol := "run:" + string(runParams) 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 // operator approved the overall direction; individual config steps
// within the window don't each need a separate yes. Destructive // within the window don't each need a separate yes. Destructive
// commands never auto-run, regardless of window. // 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) host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug)
if rerr != nil { if rerr != nil {
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error())) 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 // 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 // operator isn't asked to re-type "I confirm" for every single command
// against the thing they just confirmed. // 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) host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug)
if rerr != nil { if rerr != nil {
pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error())) 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 // assentWindowActive checks whether the operator has recently approved a plan
// in this agent's chat session. The agent sets an assent_window.agent:<uuid> // in THIS TASK's chat session. The agent sets an
// key in autonomy_settings with an expiry timestamp when chat-assent grants // assent_window.agent:<uuid>.session:<id> key in autonomy_settings with an
// a pending execution. While active, config_mutation commands auto-run // expiry timestamp when chat-assent grants a pending execution. While
// without re-approval — the operator approved the overall plan, not each step. // active, config_mutation commands auto-run without re-approval — the
func assentWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID) bool { // operator approved the overall plan, not each step. Scoped by session, not
if agentID == uuid.Nil { // just agent: with one agent:nomos entity serving every concurrent task, an
return false // 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 var expiresStr string
err := pool.QueryRow(ctx, err := pool.QueryRow(ctx,
"SELECT value FROM autonomy_settings WHERE key = $1", "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 { if err != nil {
return false 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- // destructiveWindowActive reports whether targetSlug has a live, explicitly-
// confirmed destructive grant for this agent. Key format // confirmed destructive grant for this agent WITHIN THIS SESSION/TASK. Key
// ("destructive_window.agent:<id>.target:<slug>") must match // format ("destructive_window.agent:<id>.target:<slug>.session:<id>") must
// cmd/nomos/store.go's openDestructiveWindow — both processes read/write the // match cmd/nomos/store.go's openDestructiveWindow — both processes
// same autonomy_settings row. Scoped to one target so a typed confirmation // read/write the same autonomy_settings row. Scoped to one target AND one
// for destroying container A can never be read as authorizing anything // session so a typed confirmation for destroying container A in task X can
// against container B. // never be read as authorizing anything against container A from a
func destructiveWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID, targetSlug string) bool { // different, concurrently-running task Y.
if agentID == uuid.Nil || targetSlug == "" { func destructiveWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID, targetSlug, sessionID string) bool {
if agentID == uuid.Nil || targetSlug == "" || sessionID == "" {
return false return false
} }
var expiresStr string var expiresStr string
err := pool.QueryRow(ctx, err := pool.QueryRow(ctx,
"SELECT value FROM autonomy_settings WHERE key = $1", "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 { if err != nil {
return false return false
} }