diff --git a/backend/internal/adapters/agent/activitystate/activitystate.go b/backend/internal/adapters/agent/activitystate/activitystate.go index 3c5375d9a5..d3eaceaaf7 100644 --- a/backend/internal/adapters/agent/activitystate/activitystate.go +++ b/backend/internal/adapters/agent/activitystate/activitystate.go @@ -16,6 +16,11 @@ import "github.com/aoagents/agent-orchestrator/backend/internal/domain" // - session-start / user-prompt-submit → active // - stop → idle // - permission-request → waiting_input +// +// permission-request maps to waiting_input, not blocked: none of the sharing +// adapters install the pre/post-tool-use trio, so a blocked state could never +// be cleared before the turn ends. waiting_input still suppresses automated +// nudges (NeedsInput) while leaving user-initiated sends deliverable. func StandardDeriveActivityState(event string, _ []byte) (domain.ActivityState, bool) { switch event { case "session-start": diff --git a/backend/internal/adapters/agent/autohand/autohand_test.go b/backend/internal/adapters/agent/autohand/autohand_test.go index 9387bd3e8e..3c682df609 100644 --- a/backend/internal/adapters/agent/autohand/autohand_test.go +++ b/backend/internal/adapters/agent/autohand/autohand_test.go @@ -352,7 +352,7 @@ func TestDeriveActivityState(t *testing.T) { {"session start -> active", "session-start", domain.ActivityActive, true}, {"user prompt -> active", "user-prompt-submit", domain.ActivityActive, true}, {"stop -> idle", "stop", domain.ActivityIdle, true}, - {"permission request -> waiting input", "permission-request", domain.ActivityWaitingInput, true}, + {"permission request -> waiting_input", "permission-request", domain.ActivityWaitingInput, true}, {"unknown event -> no signal", "frobnicate", "", false}, } for _, tt := range tests { diff --git a/backend/internal/adapters/agent/claudecode/activity.go b/backend/internal/adapters/agent/claudecode/activity.go index d6fabcf352..e7ae1ba4ad 100644 --- a/backend/internal/adapters/agent/claudecode/activity.go +++ b/backend/internal/adapters/agent/claudecode/activity.go @@ -20,6 +20,22 @@ func DeriveActivityState(event string, payload []byte) (domain.ActivityState, bo switch event { case "user-prompt-submit": return domain.ActivityActive, true + case "pre-tool-use", "post-tool-use", "post-tool-use-failure": + // The agent is executing tools — active. These signals carry + // tool_name/tool_use_id on the wire, and lifecycle applies them under + // a precedence rule: they never demote a sticky state, EXCEPT the + // post of the exact tool whose permission dialog blocked the session + // (approval means the tool ran — the earliest observable "the + // decision was resolved" signal, since answering a dialog fires no + // hook of its own). Without the rule, parallel-subagent traffic would + // clear a live blocked — the failure that reverted the naive mapping + // in PR #5's review. + return domain.ActivityActive, true + case "permission-request": + // Fires when a permission dialog appears — earlier and richer than + // Notification(permission_prompt): the payload names the blocking + // tool, which lifecycle snapshots for the correlated clear above. + return domain.ActivityBlocked, true case "stop": // End of a turn (including a user interrupt): the agent is idle but // alive (not exited). A following Notification(idle_prompt) also maps to @@ -35,23 +51,37 @@ func DeriveActivityState(event string, payload []byte) (domain.ActivityState, bo } } -// notificationState reports waiting_input only when the agent is genuinely -// blocked on the user: a pending tool-permission prompt (permission_prompt). -// idle_prompt means the agent finished its turn and is sitting idle at the -// prompt awaiting the next instruction — that is Idle, not a blocking request, -// so a stop/interrupt reads Idle rather than "Input Needed". Other types -// (auth_success, elicitation_*) carry no activity meaning, as does a malformed -// payload. +// notificationState splits the notification types that mean "the agent is +// paused on the user" by what unblocks them: +// - idle_prompt: the agent finished its turn and sits idle at the prompt +// awaiting the next instruction — that is Idle, not a blocking request, +// so a stop/interrupt reads Idle rather than "Input Needed". +// - agent_needs_input: a request for user input that carries no tool +// identity — waiting_input (automated nudges stay suppressed via +// NeedsInput, user sends deliver). It must NOT map to blocked: without a +// tool to correlate, the block could only lift at a turn boundary, +// rejecting user sends long after the question was answered. +// - permission_prompt: a pending permission decision (blocked — a stray +// Enter could answer the dialog). It duplicates the earlier +// permission-request hook, whose payload names the blocking tool for the +// correlated clear. +// - agent_completed (fired by `claude agents` background sessions, +// CLI 2.1.198+): the turn finished — Stop semantics, idle but alive. +// +// Other types (auth_success, elicitation_*) carry no activity meaning, as +// does a malformed payload. func notificationState(payload []byte) (domain.ActivityState, bool) { var p struct { NotificationType string `json:"notification_type"` } _ = json.Unmarshal(payload, &p) switch p.NotificationType { - case "permission_prompt": - return domain.ActivityWaitingInput, true - case "idle_prompt": + case "idle_prompt", "agent_completed": return domain.ActivityIdle, true + case "agent_needs_input": + return domain.ActivityWaitingInput, true + case "permission_prompt": + return domain.ActivityBlocked, true default: return "", false } diff --git a/backend/internal/adapters/agent/claudecode/activity_test.go b/backend/internal/adapters/agent/claudecode/activity_test.go index bd8cb881b8..2864c99664 100644 --- a/backend/internal/adapters/agent/claudecode/activity_test.go +++ b/backend/internal/adapters/agent/claudecode/activity_test.go @@ -15,9 +15,17 @@ func TestDeriveActivityState(t *testing.T) { wantOK bool }{ {"user prompt -> active", "user-prompt-submit", `{}`, domain.ActivityActive, true}, + // The tool-use trio reports active; lifecycle's precedence rule (not + // this deriver) decides whether it may demote a sticky state. + {"pre-tool-use -> active", "pre-tool-use", `{"tool_name":"Bash","tool_use_id":"toolu_1"}`, domain.ActivityActive, true}, + {"post-tool-use -> active", "post-tool-use", `{"tool_name":"Bash","tool_use_id":"toolu_1"}`, domain.ActivityActive, true}, + {"post-tool-use-failure -> active", "post-tool-use-failure", `{"tool_name":"Bash","tool_use_id":"toolu_1"}`, domain.ActivityActive, true}, + {"permission-request -> blocked", "permission-request", `{"tool_name":"Bash"}`, domain.ActivityBlocked, true}, {"stop -> idle", "stop", `{}`, domain.ActivityIdle, true}, {"notification idle_prompt -> idle", "notification", `{"notification_type":"idle_prompt"}`, domain.ActivityIdle, true}, - {"notification permission_prompt -> waiting_input", "notification", `{"notification_type":"permission_prompt"}`, domain.ActivityWaitingInput, true}, + {"notification permission_prompt -> blocked", "notification", `{"notification_type":"permission_prompt"}`, domain.ActivityBlocked, true}, + {"notification agent_needs_input -> waiting_input", "notification", `{"notification_type":"agent_needs_input"}`, domain.ActivityWaitingInput, true}, + {"notification agent_completed -> idle", "notification", `{"notification_type":"agent_completed"}`, domain.ActivityIdle, true}, {"notification auth_success -> no signal", "notification", `{"notification_type":"auth_success"}`, "", false}, {"notification empty type -> no signal", "notification", `{}`, "", false}, {"notification malformed payload -> no signal", "notification", `not json`, "", false}, diff --git a/backend/internal/adapters/agent/claudecode/claudecode.go b/backend/internal/adapters/agent/claudecode/claudecode.go index a803f86aad..f37af34cd1 100644 --- a/backend/internal/adapters/agent/claudecode/claudecode.go +++ b/backend/internal/adapters/agent/claudecode/claudecode.go @@ -62,6 +62,18 @@ func New() *Plugin { return &Plugin{} } +// EmitsSubmitActivity signals that Claude Code fires a user-prompt-submit hook +// under AO's launch, so Activity.State can flip to active after a prompt is +// accepted. See ports.ActivitySignaler. +func (p *Plugin) EmitsSubmitActivity() bool { return true } + +// EmitsBlockedActivity signals that Claude Code fires both pre- and post-tool +// hooks, so Activity.State can flip to blocked mid-turn on a permission dialog +// and the guarded send loop can clear it once the tool completes. Only +// claude-code (and its hook-delegators) carry this trio; see +// ports.ActivitySignaler. +func (p *Plugin) EmitsBlockedActivity() bool { return true } + var _ adapters.Adapter = (*Plugin)(nil) var _ ports.Agent = (*Plugin)(nil) var _ ports.AgentAuthChecker = (*Plugin)(nil) diff --git a/backend/internal/adapters/agent/claudecode/hooks.go b/backend/internal/adapters/agent/claudecode/hooks.go index 88019cdb36..1c780c22ee 100644 --- a/backend/internal/adapters/agent/claudecode/hooks.go +++ b/backend/internal/adapters/agent/claudecode/hooks.go @@ -19,10 +19,28 @@ const ( // its required "startup" matcher. var claudeStartupMatcher = "startup" -// claudeManagedHooks is the source of truth for the hooks AO installs. +// claudeManagedHooks is the source of truth for the hooks AO installs: +// SessionStart (under the "startup" matcher), UserPromptSubmit, the tool-use +// trio (PreToolUse, PostToolUse, PostToolUseFailure), PermissionRequest, +// Stop, Notification, and SessionEnd. They report normalized session metadata +// and activity-state signals back into AO's store (see DeriveActivityState). +// Notification and SessionEnd carry no matcher: each installs once and fires +// for every sub-type, and the handler filters on the payload's +// notification_type / reason field. The tool-use hooks also carry no matcher +// (fire for every tool): their payloads carry tool_name/tool_use_id, which +// lifecycle uses to clear a stale sticky `blocked` only when the specific +// approved tool finishes — the daemon-side precedence rule is what makes these +// signals safe against parallel-subagent traffic (the naive mapping without it +// was reverted in PR #5's review). PermissionRequest fires when a permission +// dialog appears and carries the blocking tool_name; `ao hooks` writes nothing +// to stdout, so installing it never injects a permission decision. var claudeManagedHooks = []hooksjson.HookSpec{ {Event: "SessionStart", Matcher: &claudeStartupMatcher, Command: claudeHookCommandPrefix + "session-start"}, {Event: "UserPromptSubmit", Command: claudeHookCommandPrefix + "user-prompt-submit"}, + {Event: "PreToolUse", Command: claudeHookCommandPrefix + "pre-tool-use"}, + {Event: "PostToolUse", Command: claudeHookCommandPrefix + "post-tool-use"}, + {Event: "PostToolUseFailure", Command: claudeHookCommandPrefix + "post-tool-use-failure"}, + {Event: "PermissionRequest", Command: claudeHookCommandPrefix + "permission-request"}, {Event: "Stop", Command: claudeHookCommandPrefix + "stop"}, {Event: "Notification", Command: claudeHookCommandPrefix + "notification"}, {Event: "SessionEnd", Command: claudeHookCommandPrefix + "session-end"}, diff --git a/backend/internal/adapters/agent/codex/activity.go b/backend/internal/adapters/agent/codex/activity.go index d934201e55..24e8b2c850 100644 --- a/backend/internal/adapters/agent/codex/activity.go +++ b/backend/internal/adapters/agent/codex/activity.go @@ -14,6 +14,9 @@ func DeriveActivityState(event string, _ []byte) (domain.ActivityState, bool) { case "user-prompt-submit": return domain.ActivityActive, true case "permission-request": + // waiting_input, not blocked: codex installs no pre/post-tool-use + // hooks, so a blocked state could never be cleared before the turn + // ends. waiting_input still suppresses automated nudges. return domain.ActivityWaitingInput, true case "stop": return domain.ActivityIdle, true diff --git a/backend/internal/adapters/agent/codex/activity_test.go b/backend/internal/adapters/agent/codex/activity_test.go index bf120f9cca..d7ce77d761 100644 --- a/backend/internal/adapters/agent/codex/activity_test.go +++ b/backend/internal/adapters/agent/codex/activity_test.go @@ -14,7 +14,7 @@ func TestDeriveActivityState(t *testing.T) { wantOK bool }{ {"user prompt -> active", "user-prompt-submit", domain.ActivityActive, true}, - {"permission request -> waiting input", "permission-request", domain.ActivityWaitingInput, true}, + {"permission request -> waiting_input", "permission-request", domain.ActivityWaitingInput, true}, {"stop -> idle", "stop", domain.ActivityIdle, true}, {"session start -> no signal", "session-start", "", false}, {"unknown event -> no signal", "frobnicate", "", false}, diff --git a/backend/internal/adapters/agent/codex/codex.go b/backend/internal/adapters/agent/codex/codex.go index 8985c0ec6a..7dd36f5f51 100644 --- a/backend/internal/adapters/agent/codex/codex.go +++ b/backend/internal/adapters/agent/codex/codex.go @@ -36,6 +36,17 @@ func New() *Plugin { return &Plugin{} } +// EmitsSubmitActivity signals Codex fires a user-prompt-submit hook under AO's +// launch. See ports.ActivitySignaler. +func (p *Plugin) EmitsSubmitActivity() bool { return true } + +// EmitsBlockedActivity is false: codex reports permission prompts as +// waiting_input — it installs no post-tool-use hook, so a blocked state could +// never be cleared mid-turn. confirmActive must not nudge it (an Enter could +// answer a pending decision it cannot report as blocked). See +// ports.ActivitySignaler. +func (p *Plugin) EmitsBlockedActivity() bool { return false } + var _ adapters.Adapter = (*Plugin)(nil) var _ ports.Agent = (*Plugin)(nil) var _ ports.AgentAuthChecker = (*Plugin)(nil) diff --git a/backend/internal/adapters/agent/droid/activity.go b/backend/internal/adapters/agent/droid/activity.go index 500eaaf2dd..c12caa0aa5 100644 --- a/backend/internal/adapters/agent/droid/activity.go +++ b/backend/internal/adapters/agent/droid/activity.go @@ -20,8 +20,12 @@ import ( // Droid's payload shapes differ from Claude Code's in one way that matters here: // the Notification payload carries no notification_type discriminator (it only // has a free-form message), but Droid only fires Notification when it needs a -// permission decision or has been idle awaiting input for 60s — both mean the -// agent is blocked on the user — so every Notification maps to waiting_input. +// permission decision or has been idle awaiting input for 60s. AO cannot tell +// which, so every Notification maps to waiting_input: it suppresses automated +// nudges (NeedsInput) in both cases, and blocked is reserved for harnesses that +// can clear it mid-turn via the pre/post-tool-use trio — droid installs no tool +// hooks, so a blocked state would linger until the turn boundary and reject +// user sends at a safely idle prompt. func DeriveActivityState(event string, payload []byte) (domain.ActivityState, bool) { switch event { case "user-prompt-submit": diff --git a/backend/internal/adapters/agent/droid/activity_test.go b/backend/internal/adapters/agent/droid/activity_test.go index 0581094197..1b0dee3b39 100644 --- a/backend/internal/adapters/agent/droid/activity_test.go +++ b/backend/internal/adapters/agent/droid/activity_test.go @@ -16,9 +16,10 @@ func TestDeriveActivityState(t *testing.T) { }{ {"user prompt -> active", "user-prompt-submit", `{}`, domain.ActivityActive, true}, {"stop -> idle", "stop", `{}`, domain.ActivityIdle, true}, - // Droid notifications fire only on permission-needed or 60s-idle, both of - // which mean the agent is blocked on the user — and the payload carries no - // notification_type to discriminate — so every notification is waiting_input. + // Droid notifications fire only on permission-needed or 60s-idle, and the + // payload carries no notification_type to discriminate — so every + // notification maps to waiting_input (an automated Enter must + // never answer a pending permission decision). {"notification -> waiting_input", "notification", `{"message":"Droid needs your permission"}`, domain.ActivityWaitingInput, true}, {"notification empty payload -> waiting_input", "notification", `{}`, domain.ActivityWaitingInput, true}, {"notification malformed payload -> waiting_input", "notification", `not json`, domain.ActivityWaitingInput, true}, diff --git a/backend/internal/adapters/runtime/conpty/client.go b/backend/internal/adapters/runtime/conpty/client.go index 628551a9c5..f4d21a3b6c 100644 --- a/backend/internal/adapters/runtime/conpty/client.go +++ b/backend/internal/adapters/runtime/conpty/client.go @@ -61,7 +61,12 @@ func clientSendMessage(addr, message string) error { } // Brief pause before Enter (matches TS: Enter sent as a separate frame). - time.Sleep(ptyInputEnterDelay) + // Skipped for an empty message — an Enter-only nudge has no paste to let + // settle, and the pause would only widen the guard-read→Enter window + // (mirrors the tmux runtime's enterDelay contract). + if len(runes) > 0 { + time.Sleep(ptyInputEnterDelay) + } frame, err := EncodeMessage(MsgTerminalInput, []byte("\r")) if err != nil { return err diff --git a/backend/internal/adapters/runtime/tmux/tmux.go b/backend/internal/adapters/runtime/tmux/tmux.go index 892add6b90..cb804b6463 100644 --- a/backend/internal/adapters/runtime/tmux/tmux.go +++ b/backend/internal/adapters/runtime/tmux/tmux.go @@ -23,6 +23,10 @@ import ( const ( defaultTimeout = 5 * time.Second defaultChunkBytes = 16 * 1024 + // defaultEnterDelay mirrors conpty's ptyInputEnterDelay: a pause after pasting + // a non-empty message, before the trailing Enter, so a large multiline paste + // does not absorb the Enter and leave the prompt unsubmitted (issue #2342). + defaultEnterDelay = 300 * time.Millisecond ) var sessionIDPattern = regexp.MustCompile(`^[a-zA-Z0-9_-]+$`) @@ -32,20 +36,22 @@ var getenv = os.Getenv // Options configures a tmux Runtime. Every field has a sensible default (see // New), so the zero value is usable. type Options struct { - Binary string // default "tmux" (resolved via exec.LookPath) - Shell string // default $SHELL else /bin/sh - Timeout time.Duration // default 5s - ChunkSize int // default 16*1024 + Binary string // default "tmux" (resolved via exec.LookPath) + Shell string // default $SHELL else /bin/sh + Timeout time.Duration // default 5s + ChunkSize int // default 16*1024 + EnterDelay time.Duration // pause after pasting a non-empty message before pressing Enter; default defaultEnterDelay. Conpty already does this (ptyInputEnterDelay); tmux lacked it, so a large multiline paste could absorb the trailing Enter and leave the prompt unsubmitted (issue #2342). } // Runtime runs agent sessions inside tmux sessions, driving them via the tmux // CLI. It implements ports.Runtime. type Runtime struct { - binary string - shell string - timeout time.Duration - chunkSize int - runner runner + binary string + shell string + timeout time.Duration + chunkSize int + enterDelay time.Duration + runner runner } var _ ports.Runtime = (*Runtime)(nil) @@ -90,12 +96,17 @@ func New(opts Options) *Runtime { if chunkSize <= 0 { chunkSize = defaultChunkBytes } + enterDelay := opts.EnterDelay + if enterDelay <= 0 { + enterDelay = defaultEnterDelay + } return &Runtime{ - binary: binary, - shell: shellPath, - timeout: timeout, - chunkSize: chunkSize, - runner: execRunner{}, + binary: binary, + shell: shellPath, + timeout: timeout, + chunkSize: chunkSize, + enterDelay: enterDelay, + runner: execRunner{}, } } @@ -190,7 +201,8 @@ func (r *Runtime) IsAlive(ctx context.Context, handle ports.RuntimeHandle) (bool } // SendMessage sends literal text to the session (chunked via send-keys -l) then -// presses Enter to submit. +// presses Enter to submit. An empty message presses Enter alone (the nudge +// contract on ports.AgentMessenger). // // ponytail: send-keys -l chunked is simpler than load-buffer/paste-buffer; the // ceiling is very large messages may be slower, but chunk size defaults to 16 KB @@ -200,12 +212,35 @@ func (r *Runtime) SendMessage(ctx context.Context, handle ports.RuntimeHandle, m if err != nil { return err } - for _, chunk := range chunks(message, r.chunkSize) { - if _, err := r.run(ctx, sendKeysLiteralArgs(id, chunk)...); err != nil { - return fmt.Errorf("tmux runtime: send message %s: %w", id, err) + enterCtx := ctx + if message != "" { + for _, chunk := range chunks(message, r.chunkSize) { + if _, err := r.run(ctx, sendKeysLiteralArgs(id, chunk)...); err != nil { + return fmt.Errorf("tmux runtime: send message %s: %w", id, err) + } + } + // Give the target TUI a moment to accept the pasted text before the + // trailing Enter, mirroring conpty's ptyInputEnterDelay. Without it a + // large multiline paste can absorb the Enter and leave the prompt + // unsubmitted (issue #2342). Empty-message nudges skip this — there is + // no paste ahead of a catch-up Enter. + // + // From here on the chunks are already in the pane, so the pause and + // the Enter are detached from the caller's cancellation (bounded by + // their own timeout instead): abandoning mid-pause would strand an + // unsubmitted draft that a retried send would then double-paste. + var cancel context.CancelFunc + enterCtx, cancel = context.WithTimeout(context.WithoutCancel(ctx), r.enterDelay+5*time.Second) + defer cancel() + if r.enterDelay > 0 { + select { + case <-enterCtx.Done(): + return enterCtx.Err() + case <-time.After(r.enterDelay): + } } } - if _, err := r.run(ctx, sendEnterArgs(id)...); err != nil { + if _, err := r.run(enterCtx, sendEnterArgs(id)...); err != nil { return fmt.Errorf("tmux runtime: send enter %s: %w", id, err) } return nil diff --git a/backend/internal/adapters/runtime/tmux/tmux_test.go b/backend/internal/adapters/runtime/tmux/tmux_test.go index a6ae8bf68e..3569d7dedd 100644 --- a/backend/internal/adapters/runtime/tmux/tmux_test.go +++ b/backend/internal/adapters/runtime/tmux/tmux_test.go @@ -46,6 +46,7 @@ func newTestRuntime(chunkSize int) (*Runtime, *fakeRunner) { fr := &fakeRunner{} r := New(Options{Binary: "tmux-test", Timeout: time.Second, Shell: "/bin/sh", ChunkSize: chunkSize}) r.runner = fr + r.enterDelay = 0 // tests must not pay the real 300ms pre-Enter pause return r, fr } @@ -480,6 +481,87 @@ func TestSendMessageUsesLiteralFlag(t *testing.T) { } } +// TestSendMessageDelaysBeforeEnter verifies the pre-Enter pause (mirroring +// conpty's ptyInputEnterDelay) fires only for a non-empty message: a large +// multiline paste needs time to settle before the trailing Enter, or the Enter +// is absorbed and the prompt is left unsubmitted (issue #2342). An empty +// (nudge) message skips the pause — there is no paste ahead of a catch-up Enter. +func TestSendMessageDelaysBeforeEnter(t *testing.T) { + // enterDelay=0 (the test default) => no pause: SendMessage is near-instant. + r0, _ := newTestRuntime(0) + r0.enterDelay = 0 + start := time.Now() + if err := r0.SendMessage(context.Background(), ports.RuntimeHandle{ID: "sess-1"}, "hi"); err != nil { + t.Fatalf("SendMessage (no delay): %v", err) + } + if dt := time.Since(start); dt > 50*time.Millisecond { + t.Fatalf("SendMessage with enterDelay=0 took %s; want no real pause", dt) + } + + // enterDelay>0 => SendMessage blocks at least enterDelay before Enter, but + // only for a non-empty message. + r, fr := newTestRuntime(0) + r.enterDelay = 30 * time.Millisecond + start = time.Now() + if err := r.SendMessage(context.Background(), ports.RuntimeHandle{ID: "sess-1"}, "hello"); err != nil { + t.Fatalf("SendMessage: %v", err) + } + if dt := time.Since(start); dt < r.enterDelay { + t.Fatalf("SendMessage took %s, want >= %s pre-Enter pause", dt, r.enterDelay) + } + // Non-empty message still ends with the literal chunks then Enter. + if len(fr.calls) != 2 { + t.Fatalf("calls = %d, want 2 (chunk + Enter)", len(fr.calls)) + } + if got, want := fr.calls[1].args, sendEnterArgs("sess-1"); !reflect.DeepEqual(got, want) { + t.Fatalf("Enter args = %#v, want %#v", got, want) + } + + // Empty (nudge) message: no paste, no pause — even with enterDelay set. + rNudge, frNudge := newTestRuntime(0) + rNudge.enterDelay = 30 * time.Millisecond + start = time.Now() + if err := rNudge.SendMessage(context.Background(), ports.RuntimeHandle{ID: "sess-1"}, ""); err != nil { + t.Fatalf("SendMessage (nudge): %v", err) + } + if dt := time.Since(start); dt > 50*time.Millisecond { + t.Fatalf("nudge SendMessage took %s; want no pause for empty message", dt) + } + // Empty message is Enter-only: no send-keys -l call, just Enter. + if len(frNudge.calls) != 1 { + t.Fatalf("nudge calls = %d, want 1 (Enter only)", len(frNudge.calls)) + } + if got, want := frNudge.calls[0].args, sendEnterArgs("sess-1"); !reflect.DeepEqual(got, want) { + t.Fatalf("nudge Enter args = %#v, want %#v", got, want) + } +} + +// TestSendMessageEnterSurvivesCallerCancel pins the detached-Enter contract: +// once the chunks are pasted, a caller cancellation landing in the pre-Enter +// pause must NOT abandon the send — the pasted draft would sit unsubmitted and +// a retried send would double-paste. The pause and Enter run on a context +// detached from the caller's, so SendMessage completes (chunks then Enter). +func TestSendMessageEnterSurvivesCallerCancel(t *testing.T) { + r, fr := newTestRuntime(0) + // A pause long enough that the 50ms-delayed cancel deterministically lands + // inside it (the chunk send is near-instant against the fake runner). + r.enterDelay = 200 * time.Millisecond + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + timer := time.AfterFunc(50*time.Millisecond, cancel) + defer timer.Stop() + + if err := r.SendMessage(ctx, ports.RuntimeHandle{ID: "sess-1"}, "hello"); err != nil { + t.Fatalf("SendMessage cancelled mid-pause: %v (Enter must run detached)", err) + } + if len(fr.calls) != 2 { + t.Fatalf("calls = %d, want 2 (chunk + Enter despite the caller cancel after the paste)", len(fr.calls)) + } + if got, want := fr.calls[1].args, sendEnterArgs("sess-1"); !reflect.DeepEqual(got, want) { + t.Fatalf("Enter args = %#v, want %#v", got, want) + } +} + // -- GetOutput tests -- func TestGetOutputValidatesLines(t *testing.T) { diff --git a/backend/internal/cli/hooks.go b/backend/internal/cli/hooks.go index 2ad06e00b6..4e3508477f 100644 --- a/backend/internal/cli/hooks.go +++ b/backend/internal/cli/hooks.go @@ -2,6 +2,7 @@ package cli import ( "context" + "encoding/json" "fmt" "io" "net/url" @@ -34,9 +35,41 @@ const ( // setActivityAPIRequest mirrors the daemon's SetActivityRequest body for // POST /api/v1/sessions/{id}/activity. The CLI keeps its own copy so it need -// not import httpd. +// not import httpd. Event carries the AO hook sub-command that produced the +// state; ToolName/ToolUseID are the tool-use correlation facts lifted from the +// native payload when present. All three are optional: an old daemon decodes +// the body leniently and simply ignores them. type setActivityAPIRequest struct { - State string `json:"state"` + State string `json:"state"` + Event string `json:"event,omitempty"` + ToolName string `json:"toolName,omitempty"` + ToolUseID string `json:"toolUseId,omitempty"` +} + +// maxActivityMetaLen caps the correlation fields lifted from a native hook +// payload before they go on the wire — they are ids/names, anything longer is +// garbage and gets dropped rather than truncated (a truncated id would never +// match its pre/post counterpart). +const maxActivityMetaLen = 256 + +// activityMeta extracts the tool-use correlation facts from a native hook +// payload. The field names are shared vocabulary across agent CLIs that emit +// them (claude-code's PreToolUse/PostToolUse/PostToolUseFailure and +// PermissionRequest payloads); adapters whose payloads lack them yield empty +// strings and the signal degrades to today's state-only form. +func activityMeta(payload []byte) (toolName, toolUseID string) { + var p struct { + ToolName string `json:"tool_name"` + ToolUseID string `json:"tool_use_id"` + } + _ = json.Unmarshal(payload, &p) + if len(p.ToolName) > maxActivityMetaLen { + p.ToolName = "" + } + if len(p.ToolUseID) > maxActivityMetaLen { + p.ToolUseID = "" + } + return p.ToolName, p.ToolUseID } // newHooksCommand builds the hidden `ao hooks ` command that @@ -81,8 +114,9 @@ func (c *commandContext) runHook(ctx context.Context, agent, event string) error return nil } + toolName, toolUseID := activityMeta(payload) path := "sessions/" + url.PathEscape(sessionID) + "/activity" - if err := c.postJSON(ctx, path, setActivityAPIRequest{State: string(state)}, nil); err != nil { + if err := c.postJSON(ctx, path, setActivityAPIRequest{State: string(state), Event: event, ToolName: toolName, ToolUseID: toolUseID}, nil); err != nil { // Surface the failure for diagnosis, but exit 0: a failed activity // report must not disrupt the agent. c.reportHookFailure(agent, event, sessionID, err) diff --git a/backend/internal/cli/hooks_test.go b/backend/internal/cli/hooks_test.go index 91260def66..641cd57136 100644 --- a/backend/internal/cli/hooks_test.go +++ b/backend/internal/cli/hooks_test.go @@ -55,10 +55,10 @@ func capturedState(t *testing.T, capture *activityCapture) string { return req.State } -func TestHooks_NotificationReportsWaitingInput(t *testing.T) { +func TestHooks_NotificationReportsBlocked(t *testing.T) { t.Setenv("AO_SESSION_ID", "ao-7") cfg := setConfigEnv(t) - srv, capture := activityServer(t, http.StatusOK, `{"ok":true,"sessionId":"ao-7","state":"waiting_input"}`) + srv, capture := activityServer(t, http.StatusOK, `{"ok":true,"sessionId":"ao-7","state":"blocked"}`) writeRunFileFor(t, cfg, srv) _, errOut, err := executeCLI(t, Deps{ @@ -71,8 +71,8 @@ func TestHooks_NotificationReportsWaitingInput(t *testing.T) { if capture.path != "/api/v1/sessions/ao-7/activity" { t.Errorf("path = %q, want /api/v1/sessions/ao-7/activity", capture.path) } - if got := capturedState(t, capture); got != "waiting_input" { - t.Errorf("state = %q, want waiting_input", got) + if got := capturedState(t, capture); got != "blocked" { + t.Errorf("state = %q, want blocked", got) } } @@ -130,7 +130,56 @@ func TestHooks_StopReportsIdle(t *testing.T) { } } -func TestHooks_CodexPermissionRequestReportsWaitingInput(t *testing.T) { +func TestHooks_ClaudeCodePermissionRequestReportsBlocked(t *testing.T) { + // claude-code installs the pre/post-tool-use trio, so a permission-request + // blocked state can be correlated and cleared — it is the one harness that + // reports blocked. + t.Setenv("AO_SESSION_ID", "ao-7") + cfg := setConfigEnv(t) + srv, capture := activityServer(t, http.StatusOK, `{"ok":true}`) + writeRunFileFor(t, cfg, srv) + + _, _, err := executeCLI(t, Deps{ + In: strings.NewReader(`{"tool_name":"Bash"}`), + ProcessAlive: func(int) bool { return true }, + }, "hooks", "claude-code", "permission-request") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := capturedState(t, capture); got != "blocked" { + t.Errorf("state = %q, want blocked", got) + } +} + +func TestHooks_PostToolUseCarriesCorrelationFields(t *testing.T) { + // Tool-use signals must carry the event and the native tool identity so + // lifecycle can clear a stale blocked only on the approved tool's post. + t.Setenv("AO_SESSION_ID", "ao-7") + cfg := setConfigEnv(t) + srv, capture := activityServer(t, http.StatusOK, `{"ok":true}`) + writeRunFileFor(t, cfg, srv) + + _, _, err := executeCLI(t, Deps{ + In: strings.NewReader(`{"tool_name":"Bash","tool_use_id":"toolu_42","tool_response":"ok"}`), + ProcessAlive: func(int) bool { return true }, + }, "hooks", "claude-code", "post-tool-use") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + var req setActivityAPIRequest + if err := json.Unmarshal([]byte(capture.body), &req); err != nil { + t.Fatalf("decode body: %v\nbody=%s", err, capture.body) + } + want := setActivityAPIRequest{State: "active", Event: "post-tool-use", ToolName: "Bash", ToolUseID: "toolu_42"} + if req != want { + t.Errorf("body = %+v, want %+v", req, want) + } +} + +func TestHooks_EventWithoutToolIdentityOmitsIt(t *testing.T) { + // Adapters whose payloads carry no tool fields (codex permission-request + // payload here has tool_name only) still tag the event; missing identity + // fields stay empty rather than inventing values. t.Setenv("AO_SESSION_ID", "ao-7") cfg := setConfigEnv(t) srv, capture := activityServer(t, http.StatusOK, `{"ok":true}`) @@ -143,8 +192,13 @@ func TestHooks_CodexPermissionRequestReportsWaitingInput(t *testing.T) { if err != nil { t.Fatalf("unexpected error: %v", err) } - if got := capturedState(t, capture); got != "waiting_input" { - t.Errorf("state = %q, want waiting_input", got) + var req setActivityAPIRequest + if err := json.Unmarshal([]byte(capture.body), &req); err != nil { + t.Fatalf("decode body: %v\nbody=%s", err, capture.body) + } + want := setActivityAPIRequest{State: "waiting_input", Event: "permission-request", ToolName: "Bash", ToolUseID: ""} + if req != want { + t.Errorf("body = %+v, want %+v", req, want) } } diff --git a/backend/internal/domain/activity.go b/backend/internal/domain/activity.go index f4a5126966..9992305ebd 100644 --- a/backend/internal/domain/activity.go +++ b/backend/internal/domain/activity.go @@ -6,18 +6,37 @@ import "time" // callbacks (see docs/agent/README.md), not inferred from transcript/JSONL type ActivityState string -// Activity states. WaitingInput is sticky (see IsSticky). +// Activity states. WaitingInput and Blocked are sticky (see IsSticky). +// +// WaitingInput and Blocked both mean "paused on the user" but demand opposite +// automation: waiting_input is an agent at an empty prompt awaiting its next +// INSTRUCTION (safe to message or nudge), while blocked is an agent stopped on +// a pending DECISION — a tool-permission or approval dialog — where a stray +// keystroke could answer the dialog on the user's behalf. Automated senders +// must never inject input into a blocked session. (Not to be confused with the +// PR-stack Blocked flag in the status read model; blocked here predates it — +// the state existed in the original activity model and returns with the +// permission-prompt producers.) const ( ActivityActive ActivityState = "active" ActivityIdle ActivityState = "idle" ActivityWaitingInput ActivityState = "waiting_input" + ActivityBlocked ActivityState = "blocked" ActivityExited ActivityState = "exited" ) // IsSticky reports whether an activity state must NOT be aged/demoted by the // passage of time (a paused agent is still paused until a new signal says so). func (a ActivityState) IsSticky() bool { - return a == ActivityWaitingInput + return a == ActivityWaitingInput || a == ActivityBlocked +} + +// NeedsInput reports whether the agent is paused on the user — waiting for the +// next instruction (waiting_input) or blocked on a decision (blocked). Both +// render as the needs_input session status. Distinct from IsSticky: stickiness +// is about time-demotion, NeedsInput about the user being the unblocker. +func (a ActivityState) NeedsInput() bool { + return a == ActivityWaitingInput || a == ActivityBlocked } // Activity captures the persisted activity reading: the state and when it was diff --git a/backend/internal/domain/activity_test.go b/backend/internal/domain/activity_test.go new file mode 100644 index 0000000000..ac1e2bbeb3 --- /dev/null +++ b/backend/internal/domain/activity_test.go @@ -0,0 +1,31 @@ +package domain + +import "testing" + +// TestActivityState_StickyAndNeedsInput pins the two independent state +// families: sticky states must survive the passage of time, and needs-input +// states are those where the user is the unblocker (waiting_input = awaiting +// the next instruction, blocked = pending permission/approval decision). +func TestActivityState_StickyAndNeedsInput(t *testing.T) { + tests := []struct { + state ActivityState + sticky bool + needsInput bool + }{ + {ActivityActive, false, false}, + {ActivityIdle, false, false}, + {ActivityWaitingInput, true, true}, + {ActivityBlocked, true, true}, + {ActivityExited, false, false}, + } + for _, tt := range tests { + t.Run(string(tt.state), func(t *testing.T) { + if got := tt.state.IsSticky(); got != tt.sticky { + t.Errorf("IsSticky() = %v, want %v", got, tt.sticky) + } + if got := tt.state.NeedsInput(); got != tt.needsInput { + t.Errorf("NeedsInput() = %v, want %v", got, tt.needsInput) + } + }) + } +} diff --git a/backend/internal/httpd/apispec/openapi.yaml b/backend/internal/httpd/apispec/openapi.yaml index 1e1ed9db23..4d82e7b44e 100644 --- a/backend/internal/httpd/apispec/openapi.yaml +++ b/backend/internal/httpd/apispec/openapi.yaml @@ -1617,6 +1617,12 @@ paths: schema: $ref: '#/components/schemas/APIError' description: Not Found + "409": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Conflict "500": content: application/json: @@ -2719,14 +2725,24 @@ components: type: object SetActivityRequest: properties: + event: + description: AO hook sub-command that produced this state (e.g. post-tool-use). + type: string state: description: Agent activity state reported by an agent hook. enum: - active - idle - waiting_input + - blocked - exited type: string + toolName: + description: Native tool name, for tool-use hook events. + type: string + toolUseId: + description: Native tool-use id, for tool-use hook events. + type: string required: - state type: object diff --git a/backend/internal/httpd/apispec/specgen/build.go b/backend/internal/httpd/apispec/specgen/build.go index df8d80692e..50c418d6fe 100644 --- a/backend/internal/httpd/apispec/specgen/build.go +++ b/backend/internal/httpd/apispec/specgen/build.go @@ -760,6 +760,10 @@ func sessionOperations() []operation { {http.StatusOK, controllers.SendSessionMessageResponse{}}, {http.StatusBadRequest, envelope.APIError{}}, {http.StatusNotFound, envelope.APIError{}}, + // Conflict: the session is terminated, or paused on a permission + // decision (SESSION_AWAITING_DECISION) — the guarded send refuses + // to paste into a pending dialog. + {http.StatusConflict, envelope.APIError{}}, {http.StatusInternalServerError, envelope.APIError{}}, }, }, diff --git a/backend/internal/httpd/controllers/dto.go b/backend/internal/httpd/controllers/dto.go index 81ac195ed9..ae986b004d 100644 --- a/backend/internal/httpd/controllers/dto.go +++ b/backend/internal/httpd/controllers/dto.go @@ -407,8 +407,17 @@ type ClaimPRResponse struct { } // SetActivityRequest is the body of POST /api/v1/sessions/{sessionId}/activity. +// Event/ToolName/ToolUseID are optional correlation facts: which AO hook +// sub-command produced the state and, for tool-use hooks, which tool call it +// concerns. Lifecycle uses them to clear a stale blocked state only when the +// specific approved tool finishes. Absent on old CLIs and on adapters whose +// payloads carry no tool identity — the signal then keeps its plain +// state-only semantics. type SetActivityRequest struct { - State string `json:"state" enum:"active,idle,waiting_input,exited" description:"Agent activity state reported by an agent hook."` + State string `json:"state" enum:"active,idle,waiting_input,blocked,exited" description:"Agent activity state reported by an agent hook."` + Event string `json:"event,omitempty" description:"AO hook sub-command that produced this state (e.g. post-tool-use)."` + ToolName string `json:"toolName,omitempty" description:"Native tool name, for tool-use hook events."` + ToolUseID string `json:"toolUseId,omitempty" description:"Native tool-use id, for tool-use hook events."` } // SetActivityResponse is the body of POST /api/v1/sessions/{sessionId}/activity. diff --git a/backend/internal/httpd/controllers/sessions.go b/backend/internal/httpd/controllers/sessions.go index 8b61fe3914..9e2977eb94 100644 --- a/backend/internal/httpd/controllers/sessions.go +++ b/backend/internal/httpd/controllers/sessions.go @@ -460,12 +460,24 @@ func (c *SessionsController) activity(w http.ResponseWriter, r *http.Request) { } state := domain.ActivityState(in.State) switch state { - case domain.ActivityActive, domain.ActivityIdle, domain.ActivityWaitingInput, domain.ActivityExited: + case domain.ActivityActive, domain.ActivityIdle, domain.ActivityWaitingInput, domain.ActivityBlocked, domain.ActivityExited: default: envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "INVALID_ACTIVITY_STATE", "Unknown activity state", nil) return } - if err := c.Activity.ApplyActivitySignal(r.Context(), sessionID(r), ports.ActivitySignal{Valid: true, State: state}); err != nil { + // The correlation fields ride the same lenient decode: absent on old CLIs. + // They are externally-supplied strings headed for logs and in-memory maps, + // so sanitize control chars and cap their length (a truncated id could + // never match its pre/post counterpart, so overlong values are dropped by + // the CLI; the cap here is defense against non-AO callers). + sig := ports.ActivitySignal{ + Valid: true, + State: state, + Event: capActivityMeta(domain.SanitizeControlChars(in.Event)), + ToolName: capActivityMeta(domain.SanitizeControlChars(in.ToolName)), + ToolUseID: capActivityMeta(domain.SanitizeControlChars(in.ToolUseID)), + } + if err := c.Activity.ApplyActivitySignal(r.Context(), sessionID(r), sig); err != nil { if errors.Is(err, ports.ErrSessionNotFound) { envelope.WriteAPIError(w, r, http.StatusNotFound, "not_found", "SESSION_NOT_FOUND", "Unknown session", nil) return @@ -476,6 +488,16 @@ func (c *SessionsController) activity(w http.ResponseWriter, r *http.Request) { envelope.WriteJSON(w, http.StatusOK, SetActivityResponse{OK: true, SessionID: sessionID(r), State: in.State}) } +// capActivityMeta bounds an optional activity correlation string; overlong +// values are dropped, not truncated (see the comment at its call site). +func capActivityMeta(v string) string { + const maxLen = 256 + if len(v) > maxLen { + return "" + } + return v +} + func (c *SessionsController) spawnOrchestrator(w http.ResponseWriter, r *http.Request) { if c.Svc == nil { apispec.NotImplemented(w, r, "POST", "/api/v1/orchestrators") diff --git a/backend/internal/httpd/controllers/sessions_activity_test.go b/backend/internal/httpd/controllers/sessions_activity_test.go index 5d7b66943e..67f93c2a0d 100644 --- a/backend/internal/httpd/controllers/sessions_activity_test.go +++ b/backend/internal/httpd/controllers/sessions_activity_test.go @@ -66,6 +66,61 @@ func TestSessionsAPI_ActivityAppliesSignal(t *testing.T) { } } +func TestSessionsAPI_ActivityAcceptsBlocked(t *testing.T) { + rec := &fakeActivityRecorder{} + srv := newActivityTestServer(t, rec) + + body, status, _ := doRequest(t, srv, "POST", "/api/v1/sessions/ao-1/activity", `{"state":"blocked"}`) + if status != http.StatusOK { + t.Fatalf("activity = %d, want 200; body=%s", status, body) + } + if !rec.gotSignal.Valid || rec.gotSignal.State != domain.ActivityBlocked { + t.Fatalf("recorder signal = %#v", rec.gotSignal) + } +} + +func TestSessionsAPI_ActivityThreadsCorrelationFields(t *testing.T) { + // The optional correlation fields ride into the signal (sanitized); a + // body without them (old CLIs) keeps producing a plain state-only signal, + // which the other tests in this file pin. + rec := &fakeActivityRecorder{} + srv := newActivityTestServer(t, rec) + + body, status, _ := doRequest(t, srv, "POST", "/api/v1/sessions/ao-1/activity", + `{"state":"active","event":"post-tool-use","toolName":"Bash","toolUseId":"toolu_42"}`) + if status != http.StatusOK { + t.Fatalf("activity = %d, want 200; body=%s", status, body) + } + want := ports.ActivitySignal{Valid: true, State: domain.ActivityActive, Event: "post-tool-use", ToolName: "Bash", ToolUseID: "toolu_42"} + if rec.gotSignal != want { + t.Fatalf("recorder signal = %#v, want %#v", rec.gotSignal, want) + } +} + +func TestSessionsAPI_ActivityCapsOverlongCorrelationFields(t *testing.T) { + // Overlong values are dropped, not truncated: a truncated id could never + // match its pre/post counterpart, so an empty value (fail-safe: no + // correlated clear) is strictly better. + rec := &fakeActivityRecorder{} + srv := newActivityTestServer(t, rec) + + long := make([]byte, 300) + for i := range long { + long[i] = 'a' + } + body, status, _ := doRequest(t, srv, "POST", "/api/v1/sessions/ao-1/activity", + `{"state":"active","event":"post-tool-use","toolUseId":"`+string(long)+`"}`) + if status != http.StatusOK { + t.Fatalf("activity = %d, want 200; body=%s", status, body) + } + if rec.gotSignal.ToolUseID != "" { + t.Fatalf("overlong toolUseId not dropped: %q", rec.gotSignal.ToolUseID) + } + if rec.gotSignal.Event != "post-tool-use" { + t.Fatalf("in-bounds event dropped: %#v", rec.gotSignal) + } +} + func TestSessionsAPI_ActivityRejectsUnknownState(t *testing.T) { rec := &fakeActivityRecorder{} srv := newActivityTestServer(t, rec) diff --git a/backend/internal/lifecycle/manager.go b/backend/internal/lifecycle/manager.go index 6212d8eb79..f3aec7b677 100644 --- a/backend/internal/lifecycle/manager.go +++ b/backend/internal/lifecycle/manager.go @@ -13,6 +13,7 @@ import ( "github.com/aoagents/agent-orchestrator/backend/internal/domain" "github.com/aoagents/agent-orchestrator/backend/internal/ports" + "github.com/aoagents/agent-orchestrator/backend/internal/sessionguard" ) type sessionStore interface { @@ -50,8 +51,11 @@ func WithTelemetry(sink ports.EventSink) Option { // Manager reduces runtime, activity, spawn, and termination observations into durable session facts. // It also owns agent nudges caused by PR observations, including merge-conflict, CI-failure, and review-feedback prompts. type Manager struct { - store sessionStore - messenger ports.AgentMessenger + store sessionStore + // guard is the shared pane-write primitive every reaction nudge goes + // through (see sessionguard). Nil when no messenger was wired: reaction + // nudges become no-ops but the reducer still runs. + guard *sessionguard.Guard notifications notificationSink mu sync.Mutex @@ -59,6 +63,9 @@ type Manager struct { clock func() time.Time react reactionState telemetry ports.EventSink + // flights tracks, per session, the in-flight tool executions and the + // pending permission dialog's identity (see toolFlight). Guarded by mu. + flights map[domain.SessionID]*toolFlight } // New builds a Lifecycle Manager over the session store it writes and the messenger it uses for agent nudges. @@ -68,7 +75,10 @@ func New(store sessionStore, messenger ports.AgentMessenger, opts ...Option) *Ma // `ao session get` showing created in UTC but updated in local time. A // WithClock option may still override this in tests. clock := func() time.Time { return time.Now().UTC() } - m := &Manager{store: store, messenger: messenger, window: defaultRecentActivityWindow, clock: clock, react: newReactionState()} + m := &Manager{store: store, window: defaultRecentActivityWindow, clock: clock, react: newReactionState(), flights: map[domain.SessionID]*toolFlight{}} + if messenger != nil { + m.guard = sessionguard.New(store, messenger, nil) + } for _, opt := range opts { opt(m) } @@ -105,6 +115,12 @@ func (m *Manager) ApplyRuntimeObservation(ctx context.Context, id domain.Session next := cur next.IsTerminated = true next.Activity = domain.Activity{State: domain.ActivityExited, LastActivityAt: timeOr(f.ObservedAt, now)} + // Reaper-driven death (crash/SIGKILL) never fires a session-end hook, + // so this is the last chance to release the session's tool-flight + // state; a leaked entry would otherwise persist for the daemon's life + // (later observations return early on cur.IsTerminated). Runs under + // m.mu — mutate holds it across this callback. + delete(m.flights, id) return next, true }) } @@ -127,6 +143,17 @@ func (m *Manager) ApplyActivitySignal(ctx context.Context, id domain.SessionID, } now := m.clock() if rec.IsTerminated { + delete(m.flights, id) + m.mu.Unlock() + return nil + } + // Event-tagged signals fold through the session's tool-flight state first: + // they may be suppressed (state write skipped) by the blocked-precedence + // rule, while their tracking side effects still land. Untagged signals + // (old CLIs, adapters without tool identity) pass through untouched — + // last-writer-wins, exactly as before. + s = m.applyToolPrecedenceLocked(id, rec.Activity.State, s) + if !s.Valid { m.mu.Unlock() return nil } @@ -155,7 +182,10 @@ func (m *Manager) ApplyActivitySignal(ctx context.Context, id domain.SessionID, m.mu.Unlock() return err } - if rec.Activity.State != domain.ActivityWaitingInput && next.Activity.State == domain.ActivityWaitingInput && !next.IsTerminated { + // Transition into the needs-input family (waiting_input or blocked) pings + // the user; an in-family escalation (waiting_input -> blocked) does not + // re-notify — the user was already pinged once for this pause. + if !rec.Activity.State.NeedsInput() && next.Activity.State.NeedsInput() && !next.IsTerminated { intent = &ports.NotificationIntent{ Type: domain.NotificationNeedsInput, SessionID: next.ID, @@ -173,6 +203,157 @@ func (m *Manager) ApplyActivitySignal(ctx context.Context, id domain.SessionID, return nil } +// toolFlight tracks one session's in-flight tool executions and the pending +// permission dialog's identity, so a sticky `blocked` is cleared by the post +// of the exact approved tool — and by nothing else tool-shaped. Answering a +// permission dialog fires no hook of its own, so the approved tool's +// post-tool-use is the earliest observable "the decision was resolved" +// signal; but tool hooks also fire for parallel subagents on the same +// session, whose traffic must never clear a dialog that is still on screen. +// In-memory only: a daemon restart loses it and the session degrades to +// clearing at the next turn boundary — fail-safe staleness, never a spurious +// clear. +type toolFlight struct { + // inflight maps toolUseID -> toolName for pre-tool-use signals whose post + // has not arrived yet. + inflight map[string]string + // blockedCandidate is the tool-use id of the UNIQUE in-flight tool bearing + // the dialog's tool_name when it appeared — the tool whose post proves the + // dialog was answered. Empty when no in-flight tool matched, or when the + // match was ambiguous (two same-name tools in flight: the permission + // payload carries no tool_use_id to disambiguate, so a sibling's post must + // NOT be mistaken for the approval). Either way, empty means nothing + // tool-shaped may clear the block and it lifts only at a turn boundary. + blockedCandidate string +} + +// maxInflightTools caps a session's in-flight map so lost posts cannot grow +// it without bound; hitting the cap resets the map, degrading that session to +// turn-boundary clearing (fail-safe). +const maxInflightTools = 128 + +// isToolUseEvent reports whether the AO hook event is one of the tool-use +// trio whose signals must not demote a sticky state on their own. +func isToolUseEvent(event string) bool { + return event == "pre-tool-use" || event == "post-tool-use" || event == "post-tool-use-failure" +} + +// isTurnBoundaryEvent reports the events that reliably mean the pending +// dialog is gone: a prompt cannot be submitted while a dialog holds the +// composer, and a turn cannot end (or the session exit) with one on screen. +func isTurnBoundaryEvent(event string) bool { + return event == "user-prompt-submit" || event == "stop" || event == "session-end" +} + +// applyToolPrecedenceLocked folds an event-tagged activity signal through the +// session's tool-flight state and decides whether its state write may +// proceed. Returned signal with Valid=false means "suppressed": the tracking +// side effects have landed but the state must not change. Signals without an +// Event pass through untouched — the compatibility contract for old CLIs and +// for adapters that don't tag their signals (their last-writer-wins semantics +// are pinned by tests). Caller must hold m.mu. +func (m *Manager) applyToolPrecedenceLocked(id domain.SessionID, cur domain.ActivityState, s ports.ActivitySignal) ports.ActivitySignal { + if s.Event == "" { + return s + } + suppressed := s + suppressed.Valid = false + + fl := m.flights[id] + ensure := func() *toolFlight { + if fl == nil { + fl = &toolFlight{inflight: map[string]string{}} + m.flights[id] = fl + } + return fl + } + + // Tracking side effects happen regardless of what the state decision is. + switch s.Event { + case "pre-tool-use": + if s.ToolUseID != "" { + f := ensure() + if len(f.inflight) >= maxInflightTools { + f.inflight = map[string]string{} + } + f.inflight[s.ToolUseID] = s.ToolName + } + case "post-tool-use", "post-tool-use-failure": + if fl != nil { + delete(fl.inflight, s.ToolUseID) + } + } + + switch { + case s.State == domain.ActivityBlocked: + // Entering (or re-asserting) blocked: snapshot the dialog's identity. + // permission-request carries the blocking tool_name; the Notification + // duplicate does not and must not wipe an existing snapshot. + // + // The permission hook payload does not carry the blocking tool's + // tool_use_id (only its name), so we can only identify the blocking + // tool unambiguously when EXACTLY ONE in-flight tool bears that name. + // With two same-name tools in flight (a batch of Bash calls, one of + // them the one at the dialog), a sibling's post could otherwise clear + // a still-open dialog — the exact permission-bypass this change exists + // to prevent. So we correlate only in the unique case; otherwise no + // candidate is recorded and the block clears only at a turn boundary + // (fail-closed). + f := ensure() + if s.ToolName != "" { + // Recompute from scratch: this is a fresh dialog, so any candidate + // carried from a prior one must not leak in. + f.blockedCandidate = "" + for useID, name := range f.inflight { + if name != s.ToolName { + continue + } + if f.blockedCandidate != "" { + // A second same-name tool: ambiguous, fail closed by + // leaving no candidate (only a turn boundary clears). + f.blockedCandidate = "" + break + } + f.blockedCandidate = useID + } + } + return s + + case cur == domain.ActivityBlocked: + // Paused on a decision: only a turn boundary or the correlated post + // may change the state. + switch { + case isTurnBoundaryEvent(s.Event): + delete(m.flights, id) + return s + case (s.Event == "post-tool-use" || s.Event == "post-tool-use-failure") && + fl != nil && fl.blockedCandidate != "" && s.ToolUseID == fl.blockedCandidate: + // The single unambiguous blocking tool finished: the dialog was + // answered. Clear the candidate so a later dialog in the same turn + // starts from a clean slate. + fl.blockedCandidate = "" + return s + default: + // Subagent/sibling tool traffic (including a same-name sibling when + // the block was ambiguous), notification sub-types (idle_prompt, + // agent_completed), and anything else that is not proof the dialog + // closed. + return suppressed + } + + case cur.IsSticky() && isToolUseEvent(s.Event): + // waiting_input: background tool traffic must not clear the "waiting + // on the user" marker; only an explicit user/turn signal does. + return suppressed + + default: + if isTurnBoundaryEvent(s.Event) { + delete(m.flights, id) + } + return s + } +} + func (m *Manager) waitingInputEvents(next domain.SessionRecord, prevState domain.ActivityState, prevAt, now time.Time) []ports.TelemetryEvent { if m.telemetry == nil { return nil @@ -180,7 +361,11 @@ func (m *Manager) waitingInputEvents(next domain.SessionRecord, prevState domain projectID := next.ProjectID sessionID := next.ID var events []ports.TelemetryEvent - if prevState != domain.ActivityWaitingInput && next.Activity.State == domain.ActivityWaitingInput && !next.IsTerminated { + // Entry/exit is measured on the needs-input family boundary (waiting_input + // or blocked): the event names stay waiting_input_* for dashboard + // continuity, the payload state distinguishes the two, and an in-family + // transition emits neither event so dwell covers the whole pause. + if !prevState.NeedsInput() && next.Activity.State.NeedsInput() && !next.IsTerminated { events = append(events, ports.TelemetryEvent{ Name: "ao.session.waiting_input_entered", Source: "lifecycle", @@ -193,7 +378,7 @@ func (m *Manager) waitingInputEvents(next domain.SessionRecord, prevState domain }, }) } - if prevState == domain.ActivityWaitingInput && next.Activity.State != domain.ActivityWaitingInput { + if prevState.NeedsInput() && !next.Activity.State.NeedsInput() { payload := map[string]any{ "state": string(next.Activity.State), "dwell_ms": now.Sub(prevAt).Milliseconds(), @@ -259,6 +444,7 @@ func (m *Manager) MarkTerminated(ctx context.Context, id domain.SessionID) error } cur.IsTerminated = true cur.Activity = domain.Activity{State: domain.ActivityExited, LastActivityAt: now} + delete(m.flights, id) // runs under m.mu (mutate holds it) return cur, true }) } diff --git a/backend/internal/lifecycle/manager_test.go b/backend/internal/lifecycle/manager_test.go index 8acb688dcc..4aa7f5b45f 100644 --- a/backend/internal/lifecycle/manager_test.go +++ b/backend/internal/lifecycle/manager_test.go @@ -636,6 +636,37 @@ func TestApplyReviewResultSendsAndDedupsThroughPRSignature(t *testing.T) { } } +func TestApplyReviewResultSuppressedByJITGuardIsNotDelivered(t *testing.T) { + // The worker is working at ApplyReviewResult's entry guard (read #1) but a + // permission dialog stores blocked before sendOnce's just-in-time re-read + // (read #2). The nudge must be SUPPRESSED, and the outcome must be + // ReviewDeliveryNoop — NOT Sent — so the caller does not stamp the run + // delivered and the changes-requested feedback re-fires once unblocked. + st := newFakeStore() + st.sessions["mer-1"] = working("mer-1") + bst := &blockOnNthGetStore{fakeStore: st, id: "mer-1", flipAt: 2} + msg := &fakeMessenger{} + m := New(bst, msg) + result := ReviewResult{ + RunID: "run-1", WorkerID: "mer-1", PRURL: "https://github.com/o/r/pull/1", + TargetSHA: "sha1", Verdict: domain.VerdictChangesRequested, Body: "fix the bug", + } + + outcome, err := m.ApplyReviewResult(ctx, "mer-1", result) + if err != nil { + t.Fatalf("ApplyReviewResult: %v", err) + } + if outcome != ReviewDeliveryNoop { + t.Fatalf("outcome = %q, want no_op (suppressed nudge must not be stamped delivered)", outcome) + } + if len(msg.msgs) != 0 { + t.Fatalf("nudge pasted into a session that went blocked before send: %v", msg.msgs) + } + if st.signatures[result.PRURL] != "" { + t.Fatal("suppressed nudge must not persist a sendOnce signature (it re-fires next observation)") + } +} + func TestApplyReviewBatchSendsCombinedAndDedups(t *testing.T) { st := newFakeStore() st.sessions["mer-1"] = working("mer-1") @@ -1017,6 +1048,156 @@ func TestActivity_WaitingInputSameStateDoesNotEmitNotification(t *testing.T) { } } +func TestActivity_BlockedTransitionEmitsNotification(t *testing.T) { + st := newFakeStore() + sink := &fakeNotificationSink{} + m := New(st, nil, WithNotificationSink(sink)) + now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC) + m.clock = func() time.Time { return now } + st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", DisplayName: "checkout-flow", Activity: domain.Activity{State: domain.ActivityActive, LastActivityAt: now.Add(-time.Minute)}, FirstSignalAt: now.Add(-time.Minute)} + + if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityBlocked}); err != nil { + t.Fatal(err) + } + if len(sink.intents) != 1 { + t.Fatalf("intents = %d, want 1 (blocked is a needs-input entry)", len(sink.intents)) + } + if sink.intents[0].Type != domain.NotificationNeedsInput { + t.Fatalf("intent type = %q, want needs_input", sink.intents[0].Type) + } +} + +func TestActivity_WaitingInputToBlockedDoesNotReNotify(t *testing.T) { + // waiting_input -> blocked is an in-family escalation: the user was already + // pinged once for this pause, so no second notification and no telemetry + // entry/exit pair. + st := newFakeStore() + sink := &fakeNotificationSink{} + tele := &telemetrySink{} + m := New(st, nil, WithNotificationSink(sink), WithTelemetry(tele)) + now := time.Date(2026, 7, 2, 10, 0, 0, 0, time.UTC) + m.clock = func() time.Time { return now } + st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Activity: domain.Activity{State: domain.ActivityWaitingInput, LastActivityAt: now.Add(-time.Minute)}, FirstSignalAt: now.Add(-time.Minute)} + + if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityBlocked}); err != nil { + t.Fatal(err) + } + if len(sink.intents) != 0 { + t.Fatalf("in-family escalation emitted notification: %+v", sink.intents) + } + if len(tele.events) != 0 { + t.Fatalf("in-family escalation emitted telemetry: %+v", tele.events) + } +} + +func TestActivity_BlockedEntryAndExitEmitTelemetry(t *testing.T) { + st := newFakeStore() + sink := &telemetrySink{} + m := New(st, nil, WithTelemetry(sink)) + now := time.Unix(100, 0).UTC() + m.clock = func() time.Time { return now } + st.sessions["mer-1"] = domain.SessionRecord{ + ID: "mer-1", + ProjectID: "mer", + Activity: domain.Activity{State: domain.ActivityIdle, LastActivityAt: now.Add(-time.Minute)}, + } + + if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityBlocked, Timestamp: now}); err != nil { + t.Fatal(err) + } + now = now.Add(2 * time.Second) + if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityActive, Timestamp: now}); err != nil { + t.Fatal(err) + } + + if len(sink.events) != 2 { + t.Fatalf("events = %#v, want entered/exited", sink.events) + } + if sink.events[0].Name != "ao.session.waiting_input_entered" || sink.events[1].Name != "ao.session.waiting_input_exited" { + t.Fatalf("event names = %#v (family events keep the waiting_input_* names)", []string{sink.events[0].Name, sink.events[1].Name}) + } + if got := sink.events[0].Payload["state"]; got != "blocked" { + t.Fatalf("entered payload state = %#v, want blocked", got) + } +} + +func TestSCMObservation_ReadyToMergeSuppressedWhileBlocked(t *testing.T) { + st := newFakeStore() + sink := &fakeNotificationSink{} + m := New(st, nil, WithNotificationSink(sink)) + rec := working("mer-1") + rec.Activity.State = domain.ActivityBlocked + st.sessions["mer-1"] = rec + obs := ports.SCMObservation{ + Fetched: true, + PR: ports.SCMPRObservation{URL: "https://github.com/o/r/pull/1", Number: 1}, + CI: ports.SCMCIObservation{Summary: string(domain.CIPassing)}, + Mergeability: ports.SCMMergeabilityObservation{State: string(domain.MergeMergeable)}, + } + if err := m.ApplySCMObservation(ctx, "mer-1", obs); err != nil { + t.Fatal(err) + } + if len(sink.intents) != 0 { + t.Fatalf("blocked session emitted ready notification: %+v", sink.intents) + } +} + +// blockOnNthGetStore wraps fakeStore and flips a session to ActivityBlocked on +// the Nth GetSession call, reproducing the reactions TOCTOU: the handler's +// entry guard (1st read) sees the session working, but a permission hook stores +// blocked before sendOnce's just-in-time re-read (2nd read). +type blockOnNthGetStore struct { + *fakeStore + id domain.SessionID + reads int + flipAt int +} + +func (s *blockOnNthGetStore) GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) { + s.reads++ + if s.reads == s.flipAt { + if rec, ok := s.sessions[s.id]; ok { + rec.Activity.State = domain.ActivityBlocked + s.sessions[s.id] = rec + } + } + return s.fakeStore.GetSession(ctx, id) +} + +func TestSendOnce_NoNudgeWhenBlockedAppearsBeforeSend(t *testing.T) { + // The entry guard in ApplyPRObservation reads the session working (read #1); + // a permission dialog then stores blocked before sendOnce's just-in-time + // re-read (read #2), which must suppress the paste+Enter into the dialog. + st := newFakeStore() + st.sessions["mer-1"] = working("mer-1") + bst := &blockOnNthGetStore{fakeStore: st, id: "mer-1", flipAt: 2} + msg := &fakeMessenger{} + m := New(bst, msg) + o := ports.PRObservation{Fetched: true, URL: "pr1", CI: domain.CIFailing, Checks: []ports.PRCheckObservation{{Name: "build", CommitHash: "c1", Status: domain.PRCheckFailed, LogTail: "boom"}}} + if err := m.ApplyPRObservation(ctx, "mer-1", o); err != nil { + t.Fatal(err) + } + if len(msg.msgs) != 0 { + t.Fatalf("nudge sent into a session that went blocked before send: %v", msg.msgs) + } +} + +func TestPRObservation_NudgesSuppressedWhileBlocked(t *testing.T) { + // A blocked session must not receive automated CI/review nudges: injected + // text could interact with the pending permission dialog. + m, st, msg := newManager() + rec := working("mer-1") + rec.Activity.State = domain.ActivityBlocked + st.sessions["mer-1"] = rec + o := ports.PRObservation{Fetched: true, URL: "pr1", CI: domain.CIFailing, Checks: []ports.PRCheckObservation{{Name: "build", CommitHash: "c1", Status: domain.PRCheckFailed, LogTail: "boom"}}} + if err := m.ApplyPRObservation(ctx, "mer-1", o); err != nil { + t.Fatal(err) + } + if len(msg.msgs) != 0 { + t.Fatalf("blocked session got nudged: %v", msg.msgs) + } +} + func TestActivity_TerminatedSessionDoesNotEmitNotification(t *testing.T) { st := newFakeStore() sink := &fakeNotificationSink{} diff --git a/backend/internal/lifecycle/reactions.go b/backend/internal/lifecycle/reactions.go index 7c506ccb0b..6b88b36cd0 100644 --- a/backend/internal/lifecycle/reactions.go +++ b/backend/internal/lifecycle/reactions.go @@ -12,6 +12,7 @@ import ( "github.com/aoagents/agent-orchestrator/backend/internal/domain" "github.com/aoagents/agent-orchestrator/backend/internal/ports" + "github.com/aoagents/agent-orchestrator/backend/internal/sessionguard" ) const reviewMaxNudge = 3 @@ -55,10 +56,10 @@ func (m *Manager) ApplyReviewBatch(ctx context.Context, workerID domain.SessionI if err != nil || !ok { return ReviewDeliveryNoop, err } - if rec.IsTerminated || rec.Activity.State == domain.ActivityWaitingInput { + if rec.IsTerminated || rec.Activity.State.NeedsInput() { return ReviewDeliveryNoop, nil } - if m.messenger == nil { + if m.guard == nil { return ReviewDeliveryNoop, nil } sort.Slice(results, func(i, j int) bool { @@ -88,9 +89,16 @@ func (m *Manager) ApplyReviewBatch(ctx context.Context, workerID domain.SessionI anchorPR := results[0].PRURL key := "review-batch:" + anchorPR + ":" + batchID sig := strings.Join(sigParts, "\x01") - if err := m.sendOnce(ctx, workerID, anchorPR, key, sig, msg.String(), reviewMaxNudge); err != nil { + outcome, err := m.sendOnce(ctx, workerID, anchorPR, key, sig, msg.String(), reviewMaxNudge) + if err != nil { return ReviewDeliveryNoop, err } + if outcome == sendOnceSuppressed { + // The worker went terminated/needs-input between the entry guard and the + // paste: nothing reached it, so do NOT let the caller stamp the run + // delivered — it must re-fire once the session is workable again. + return ReviewDeliveryNoop, nil + } return ReviewDeliverySent, nil } @@ -154,7 +162,7 @@ func (m *Manager) ApplyPRObservation(ctx context.Context, id domain.SessionID, o if err != nil || !ok { return err } - if rec.IsTerminated || rec.Activity.State == domain.ActivityWaitingInput { + if rec.IsTerminated || rec.Activity.State.NeedsInput() { return nil } // A single PR can trip several actionable conditions at once (failing CI, @@ -216,7 +224,7 @@ func (m *Manager) ApplyPRObservation(ctx context.Context, id domain.SessionID, o } for _, n := range nudges { - if err := m.sendOnce(ctx, id, o.URL, n.key, n.sig, n.msg, n.maxAttempts); err != nil { + if _, err := m.sendOnce(ctx, id, o.URL, n.key, n.sig, n.msg, n.maxAttempts); err != nil { return err } } @@ -234,10 +242,10 @@ func (m *Manager) ApplyReviewResult(ctx context.Context, workerID domain.Session if err != nil || !ok { return ReviewDeliveryNoop, err } - if rec.IsTerminated || rec.Activity.State == domain.ActivityWaitingInput { + if rec.IsTerminated || rec.Activity.State.NeedsInput() { return ReviewDeliveryNoop, nil } - if m.messenger == nil { + if m.guard == nil { return ReviewDeliveryNoop, nil } msg := fmt.Sprintf("[AO reviewer] AO's internal code reviewer submitted a review.\n\nPR: %s\nVerdict: %s", domain.SanitizeControlChars(r.PRURL), domain.SanitizeControlChars(string(r.Verdict))) @@ -251,10 +259,16 @@ func (m *Manager) ApplyReviewResult(ctx context.Context, workerID domain.Session } key := "review:" + r.PRURL + ":ao:" + r.RunID sig := strings.Join([]string{r.TargetSHA, r.RunID, r.GithubReviewID, r.Body}, "\x00") - err = m.sendOnce(ctx, workerID, r.PRURL, key, sig, msg, reviewMaxNudge) + outcome, err := m.sendOnce(ctx, workerID, r.PRURL, key, sig, msg, reviewMaxNudge) if err != nil { return ReviewDeliveryNoop, err } + if outcome == sendOnceSuppressed { + // Suppressed by the just-in-time guard (worker went terminated/needs- + // input): the review feedback did not reach the worker, so leave the run + // undelivered to re-fire on the next observation. + return ReviewDeliveryNoop, nil + } return ReviewDeliverySent, nil } @@ -359,7 +373,7 @@ func (m *Manager) notificationIntentForSCM(rec domain.SessionRecord, o ports.SCM base.Type = domain.NotificationPRClosedUnmerged return &base } - if rec.IsTerminated || rec.Activity.State == domain.ActivityWaitingInput || !scmObservationIsReadyToMerge(o) { + if rec.IsTerminated || rec.Activity.State.NeedsInput() || !scmObservationIsReadyToMerge(o) { return nil } base.Type = domain.NotificationReadyToMerge @@ -488,7 +502,7 @@ func (m *Manager) ApplyTrackerFacts(ctx context.Context, id domain.SessionID, o if err != nil || !ok { return err } - if rec.IsTerminated || rec.Activity.State == domain.ActivityWaitingInput { + if rec.IsTerminated || rec.Activity.State.NeedsInput() { return nil } if o.Changed.Assignee { @@ -506,7 +520,8 @@ func (m *Manager) ApplyTrackerFacts(ctx context.Context, id domain.SessionID, o // the PR-row signature load/persist is skipped, so the dedup // survives only for the lifetime of this Manager. Cross-restart // persistence ships with #35. - return m.sendOnce(ctx, id, "", "tracker-bot:"+o.Issue.URL, strings.Join(ids, ","), msg, 0) + _, err := m.sendOnce(ctx, id, "", "tracker-bot:"+o.Issue.URL, strings.Join(ids, ","), msg, 0) + return err } } return nil @@ -603,29 +618,63 @@ func reviewContent(comments []ports.PRCommentObservation) (string, string) { return strings.Join(bodies, "\n\n"), strings.Join(ids, ",") } -func (m *Manager) sendOnce(ctx context.Context, id domain.SessionID, prURL, key, sig, msg string, maxAttempts int) error { - if m.messenger == nil { - return nil +// sendOnceOutcome tells a caller whether a nudge is accounted for (actually +// sent, or already covered by dedup state) versus suppressed by the just-in-time +// session guard. It matters for review delivery: a suppressed nudge must NOT be +// stamped delivered, or the feedback is lost when the session later unblocks. +type sendOnceOutcome int + +const ( + // sendOnceAccounted: the message was sent, or a prior identical send is + // already recorded (dedup hit) or the attempt budget is spent. In every + // case the caller may treat the nudge as delivered — nothing more to do. + sendOnceAccounted sendOnceOutcome = iota + // sendOnceSuppressed: the just-in-time guard skipped the paste because the + // session is terminated or awaiting the user (blocked/waiting_input). The + // message did NOT reach the worker; the caller must not mark it delivered so + // it re-fires on the next observation once the session is workable again. + sendOnceSuppressed +) + +func (m *Manager) sendOnce(ctx context.Context, id domain.SessionID, prURL, key, sig, msg string, maxAttempts int) (sendOnceOutcome, error) { + if m.guard == nil { + return sendOnceAccounted, nil } m.react.mu.Lock() defer m.react.mu.Unlock() if prURL != "" && !m.react.loaded[prURL] { if err := m.loadPRSignaturesLocked(ctx, prURL); err != nil { - return err + return sendOnceAccounted, err } m.react.loaded[prURL] = true } if m.react.seen[key] == sig { - return nil + return sendOnceAccounted, nil } attempts := m.react.attempts[key] if maxAttempts > 0 && attempts >= maxAttempts { - return nil + return sendOnceAccounted, nil + } + // The guard re-reads the session immediately before pasting: the caller's + // NeedsInput() entry check ran before this function's dedup/persist I/O, so + // a permission hook could have stored blocked (or the session could have + // terminated) in the meantime. A suppressed write returns SUPPRESSED (not + // accounted), so a review caller won't stamp it delivered and it re-fires + // once the session is workable again. A store failure inside the guard also + // suppresses (fail closed, nothing was written); a messenger failure means + // the write was attempted and stays accounted, matching the pre-guard + // behavior. + outcome, err := m.guard.Nudge(ctx, id, msg) + if err != nil { + if outcome != sessionguard.Sent { + return sendOnceSuppressed, err + } + return sendOnceAccounted, err } - if err := m.messenger.Send(ctx, id, msg); err != nil { - return err + if outcome != sessionguard.Sent { + return sendOnceSuppressed, nil } // Order: Send → in-memory mutation → durable persist. Sending first means a // transient persist failure does NOT swallow a real send (the agent saw the @@ -637,10 +686,10 @@ func (m *Manager) sendOnce(ctx context.Context, id domain.SessionID, prURL, key, m.react.attempts[key] = attempts + 1 if prURL != "" { if err := m.persistPRSignaturesLocked(ctx, prURL); err != nil { - return err + return sendOnceAccounted, err } } - return nil + return sendOnceAccounted, nil } // loadPRSignaturesLocked merges any previously persisted reaction-dedup state diff --git a/backend/internal/lifecycle/toolflight_test.go b/backend/internal/lifecycle/toolflight_test.go new file mode 100644 index 0000000000..a3c3e782bd --- /dev/null +++ b/backend/internal/lifecycle/toolflight_test.go @@ -0,0 +1,306 @@ +package lifecycle + +import ( + "testing" + "time" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" + "github.com/aoagents/agent-orchestrator/backend/internal/ports" +) + +// sig builds an event-tagged activity signal the way the ingress does. +func sig(state domain.ActivityState, event, toolName, toolUseID string) ports.ActivitySignal { + return ports.ActivitySignal{Valid: true, State: state, Event: event, ToolName: toolName, ToolUseID: toolUseID} +} + +// seedSignaled seeds a session that already produced signals (FirstSignalAt +// stamped), so receipt stamping does not interfere with precedence assertions. +func seedSignaled(st *fakeStore, id domain.SessionID, state domain.ActivityState) { + st.sessions[id] = domain.SessionRecord{ + ID: id, ProjectID: "mer", + Activity: domain.Activity{State: state, LastActivityAt: time.Now()}, + FirstSignalAt: time.Now(), + } +} + +// mustApply applies a signal and fails the test on error. +func mustApply(t *testing.T, m *Manager, id domain.SessionID, s ports.ActivitySignal) { + t.Helper() + if err := m.ApplyActivitySignal(ctx, id, s); err != nil { + t.Fatal(err) + } +} + +func stateOf(st *fakeStore, id domain.SessionID) domain.ActivityState { + return st.sessions[id].Activity.State +} + +// blockOnDialog drives a session into blocked through the real signal path: +// the blocking tool's pre-tool-use, then permission-request naming that tool. +func blockOnDialog(t *testing.T, m *Manager, st *fakeStore, id domain.SessionID, toolName, toolUseID string) { + t.Helper() + mustApply(t, m, id, sig(domain.ActivityActive, "pre-tool-use", toolName, toolUseID)) + mustApply(t, m, id, sig(domain.ActivityBlocked, "permission-request", toolName, "")) + if got := stateOf(st, id); got != domain.ActivityBlocked { + t.Fatalf("setup: state = %q, want blocked", got) + } +} + +func TestToolPrecedence_ApprovedToolPostClearsBlocked(t *testing.T) { + // Approving the dialog fires no hook; the approved tool's own post is the + // earliest observable resolution signal and must clear blocked -> active. + m, st, _ := newManager() + seedSignaled(st, "mer-1", domain.ActivityActive) + blockOnDialog(t, m, st, "mer-1", "Bash", "toolu_1") + + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Bash", "toolu_1")) + if got := stateOf(st, "mer-1"); got != domain.ActivityActive { + t.Fatalf("state after approved tool's post = %q, want active", got) + } +} + +func TestToolPrecedence_ApprovedToolFailurePostAlsoClears(t *testing.T) { + // An approved tool that runs and FAILS still resolved the dialog. + m, st, _ := newManager() + seedSignaled(st, "mer-1", domain.ActivityActive) + blockOnDialog(t, m, st, "mer-1", "Bash", "toolu_1") + + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use-failure", "Bash", "toolu_1")) + if got := stateOf(st, "mer-1"); got != domain.ActivityActive { + t.Fatalf("state after approved tool's failure post = %q, want active", got) + } +} + +func TestToolPrecedence_SubagentTrafficDoesNotClearBlocked(t *testing.T) { + // The failure that reverted the naive mapping in PR #5's review: parallel + // subagent tool signals (same session id) land while the dialog is still + // on screen. They must be tracked but never change the state. + m, st, _ := newManager() + seedSignaled(st, "mer-1", domain.ActivityActive) + blockOnDialog(t, m, st, "mer-1", "Bash", "toolu_1") + + // A different tool starts and finishes while the dialog is pending. + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "pre-tool-use", "Read", "toolu_sub")) + if got := stateOf(st, "mer-1"); got != domain.ActivityBlocked { + t.Fatalf("state after subagent pre = %q, want blocked", got) + } + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Read", "toolu_sub")) + if got := stateOf(st, "mer-1"); got != domain.ActivityBlocked { + t.Fatalf("state after subagent post = %q, want blocked", got) + } + // The approved tool's post still clears afterwards. + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Bash", "toolu_1")) + if got := stateOf(st, "mer-1"); got != domain.ActivityActive { + t.Fatalf("state after approved post = %q, want active", got) + } +} + +func TestToolPrecedence_TurnBoundariesClearBlocked(t *testing.T) { + // A prompt cannot be submitted and a turn cannot end while a dialog holds + // the composer, so these events reliably mean the dialog is gone. + cases := []struct { + name string + s ports.ActivitySignal + want domain.ActivityState + }{ + {"user-prompt-submit", sig(domain.ActivityActive, "user-prompt-submit", "", ""), domain.ActivityActive}, + {"stop", sig(domain.ActivityIdle, "stop", "", ""), domain.ActivityIdle}, + {"session-end", sig(domain.ActivityExited, "session-end", "", ""), domain.ActivityExited}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + m, st, _ := newManager() + seedSignaled(st, "mer-1", domain.ActivityActive) + blockOnDialog(t, m, st, "mer-1", "Bash", "toolu_1") + mustApply(t, m, "mer-1", tc.s) + if got := stateOf(st, "mer-1"); got != tc.want { + t.Fatalf("state = %q, want %q", got, tc.want) + } + }) + } +} + +func TestToolPrecedence_NotificationSubtypesDoNotClearBlocked(t *testing.T) { + // agent_completed (idle) and idle_prompt (waiting_input) arrive with + // event "notification" and are no evidence the dialog closed — a + // background `claude agents` run finishing must not unmask a live dialog + // (the cycle-2 minor finding on PR #5). + m, st, _ := newManager() + seedSignaled(st, "mer-1", domain.ActivityActive) + blockOnDialog(t, m, st, "mer-1", "Bash", "toolu_1") + + mustApply(t, m, "mer-1", sig(domain.ActivityIdle, "notification", "", "")) + if got := stateOf(st, "mer-1"); got != domain.ActivityBlocked { + t.Fatalf("state after notification idle = %q, want blocked", got) + } + mustApply(t, m, "mer-1", sig(domain.ActivityWaitingInput, "notification", "", "")) + if got := stateOf(st, "mer-1"); got != domain.ActivityBlocked { + t.Fatalf("state after notification waiting_input = %q, want blocked", got) + } +} + +func TestToolPrecedence_NoCandidatesFailsSafe(t *testing.T) { + // A blocking signal that carries no tool identity (codex, a bare + // Notification, or a daemon restarted mid-dialog) yields no candidates: + // nothing tool-shaped may clear the block; the turn boundary still does. + m, st, _ := newManager() + seedSignaled(st, "mer-1", domain.ActivityActive) + // Blocked lands with NO prior pre and NO tool name (restart-equivalent). + mustApply(t, m, "mer-1", sig(domain.ActivityBlocked, "permission-request", "", "")) + + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Bash", "toolu_1")) + if got := stateOf(st, "mer-1"); got != domain.ActivityBlocked { + t.Fatalf("state after uncorrelated post = %q, want blocked (fail safe)", got) + } + mustApply(t, m, "mer-1", sig(domain.ActivityIdle, "stop", "", "")) + if got := stateOf(st, "mer-1"); got != domain.ActivityIdle { + t.Fatalf("state after stop = %q, want idle", got) + } +} + +func TestToolPrecedence_LegacySignalsKeepLastWriterWins(t *testing.T) { + // The compatibility pin: a signal WITHOUT an event (old CLIs, the 12 + // adapters that don't tag their signals) keeps today's last-writer-wins + // semantics even out of blocked — the precedence rule must not change + // behavior for anyone who doesn't opt in. + m, st, _ := newManager() + seedSignaled(st, "mer-1", domain.ActivityActive) + blockOnDialog(t, m, st, "mer-1", "Bash", "toolu_1") + + mustApply(t, m, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityActive}) + if got := stateOf(st, "mer-1"); got != domain.ActivityActive { + t.Fatalf("legacy active over blocked = %q, want active (last-writer-wins preserved)", got) + } +} + +func TestToolPrecedence_ToolEventsDoNotDemoteWaitingInput(t *testing.T) { + // waiting_input marks "the user's turn". Background subagent tool traffic + // must not clear it; an explicit user signal does. + m, st, _ := newManager() + seedSignaled(st, "mer-1", domain.ActivityWaitingInput) + + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "pre-tool-use", "Read", "toolu_bg")) + if got := stateOf(st, "mer-1"); got != domain.ActivityWaitingInput { + t.Fatalf("state after background pre = %q, want waiting_input", got) + } + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Read", "toolu_bg")) + if got := stateOf(st, "mer-1"); got != domain.ActivityWaitingInput { + t.Fatalf("state after background post = %q, want waiting_input", got) + } + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "user-prompt-submit", "", "")) + if got := stateOf(st, "mer-1"); got != domain.ActivityActive { + t.Fatalf("state after user prompt = %q, want active", got) + } +} + +func TestToolPrecedence_AmbiguousSameNameBlockFailsClosed(t *testing.T) { + // Two same-name tools in flight when the dialog appears: the permission + // payload has no tool_use_id to say WHICH one is at the dialog, so neither + // sibling's post may clear the block — that would let a paste answer a + // still-open dialog. Fail closed: the block lifts only at a turn boundary. + m, st, _ := newManager() + seedSignaled(st, "mer-1", domain.ActivityActive) + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "pre-tool-use", "Bash", "toolu_1")) + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "pre-tool-use", "Bash", "toolu_2")) + mustApply(t, m, "mer-1", sig(domain.ActivityBlocked, "permission-request", "Bash", "")) + + // Either sibling's post must NOT clear the live dialog. + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Bash", "toolu_2")) + if got := stateOf(st, "mer-1"); got != domain.ActivityBlocked { + t.Fatalf("state after ambiguous sibling post = %q, want blocked (fail closed)", got) + } + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Bash", "toolu_1")) + if got := stateOf(st, "mer-1"); got != domain.ActivityBlocked { + t.Fatalf("state after second ambiguous post = %q, want blocked (fail closed)", got) + } + // A turn boundary still lifts it. + mustApply(t, m, "mer-1", sig(domain.ActivityIdle, "stop", "", "")) + if got := stateOf(st, "mer-1"); got != domain.ActivityIdle { + t.Fatalf("state after stop = %q, want idle", got) + } +} + +func TestToolPrecedence_SequentialDialogsResetAmbiguity(t *testing.T) { + // Two dialogs in one turn without a turn boundary between them (the block + // re-asserts via a fresh permission-request). A prior ambiguous block must + // not leak into a later unique one: the re-snapshot recomputes from + // scratch, so the unique dialog's tool can still clear it. + m, st, _ := newManager() + seedSignaled(st, "mer-1", domain.ActivityActive) + // Dialog 1: two same-name Bash tools -> ambiguous, cannot clear on a post. + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "pre-tool-use", "Bash", "toolu_1")) + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "pre-tool-use", "Bash", "toolu_2")) + mustApply(t, m, "mer-1", sig(domain.ActivityBlocked, "permission-request", "Bash", "")) + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Bash", "toolu_1")) + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Bash", "toolu_2")) + if got := stateOf(st, "mer-1"); got != domain.ActivityBlocked { + t.Fatalf("dialog 1 ambiguous: state = %q, want blocked", got) + } + // Both dialog-1 tools have posted, so their ids are drained from inflight. + // Dialog 2: a single new Bash -> unique, must be clearable despite the + // stale ambiguousBlock from dialog 1. + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "pre-tool-use", "Bash", "toolu_3")) + mustApply(t, m, "mer-1", sig(domain.ActivityBlocked, "permission-request", "Bash", "")) + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Bash", "toolu_3")) + if got := stateOf(st, "mer-1"); got != domain.ActivityActive { + t.Fatalf("dialog 2 unique: state = %q, want active (ambiguity must reset)", got) + } +} + +func TestToolPrecedence_UniqueSameNameToolClears(t *testing.T) { + // The unambiguous case still works: exactly one in-flight Bash when the + // dialog appears, so its post is a valid approval signal. + m, st, _ := newManager() + seedSignaled(st, "mer-1", domain.ActivityActive) + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "pre-tool-use", "Bash", "toolu_1")) + mustApply(t, m, "mer-1", sig(domain.ActivityBlocked, "permission-request", "Bash", "")) + + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Bash", "toolu_1")) + if got := stateOf(st, "mer-1"); got != domain.ActivityActive { + t.Fatalf("state after unique tool's post = %q, want active", got) + } +} + +func TestToolPrecedence_ReaperTerminationReleasesFlight(t *testing.T) { + // A crash/SIGKILL is reaped via ApplyRuntimeObservation, which fires no + // session-end hook — so it is the last chance to release the session's + // tool-flight state. A leaked entry would otherwise persist for the + // daemon's life (later observations return early on cur.IsTerminated). + m, st, _ := newManager() + // A live (non-sticky) session with stale activity so the reaper considers + // it clearly dead, plus an in-flight tool tracked in the flights map. + st.sessions["mer-1"] = domain.SessionRecord{ + ID: "mer-1", ProjectID: "mer", + Activity: domain.Activity{State: domain.ActivityActive, LastActivityAt: time.Now().Add(-2 * time.Minute)}, + FirstSignalAt: time.Now().Add(-2 * time.Minute), + } + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "pre-tool-use", "Bash", "toolu_1")) + if _, ok := m.flights["mer-1"]; !ok { + t.Fatal("setup: expected a flights entry after pre-tool-use") + } + if err := m.ApplyRuntimeObservation(ctx, "mer-1", ports.RuntimeFacts{Probe: ports.ProbeDead}); err != nil { + t.Fatal(err) + } + if !st.sessions["mer-1"].IsTerminated { + t.Fatal("reaper did not terminate the session") + } + if _, ok := m.flights["mer-1"]; ok { + t.Fatal("flights entry leaked after reaper termination") + } +} + +func TestToolPrecedence_SuppressedSignalEmitsNoNotification(t *testing.T) { + // A suppressed clear must not fan out: the session never left blocked, so + // no needs-input exit/entry telemetry or notification may fire. + st := newFakeStore() + sink := &fakeNotificationSink{} + m := New(st, nil, WithNotificationSink(sink)) + seedSignaled(st, "mer-1", domain.ActivityActive) + blockOnDialog(t, m, st, "mer-1", "Bash", "toolu_1") + entered := len(sink.intents) // the blocked entry notification + + mustApply(t, m, "mer-1", sig(domain.ActivityActive, "post-tool-use", "Read", "toolu_sub")) + if len(sink.intents) != entered { + t.Fatalf("suppressed signal emitted a notification: %d -> %d", entered, len(sink.intents)) + } +} diff --git a/backend/internal/ports/agent.go b/backend/internal/ports/agent.go index 057a38ab2f..13d2e51180 100644 --- a/backend/internal/ports/agent.go +++ b/backend/internal/ports/agent.go @@ -96,6 +96,40 @@ type AgentResolver interface { Agent(harness domain.AgentHarness) (Agent, bool) } +// ActivitySignaler is an OPTIONAL capability an Agent adapter may implement to +// describe which durable activity signals its harness actually produces under +// AO's headless launch. The Session Manager gates best-effort post-send +// confirmation on it — see the two methods. +// +// EmitsSubmitActivity reports whether the harness emits a prompt-submit signal +// (one that flips Activity.State to active). Without it the confirm loop could +// never observe active and would only burn its budget on spurious Enter nudges. +// +// EmitsBlockedActivity reports whether the harness emits a decision-pause +// signal (a permission/approval prompt that flips Activity.State to blocked) +// AND can clear that state before the turn ends — which requires the +// pre/post-tool-use trio so lifecycle can correlate the approved tool's post +// with the dialog that blocked the session. The Enter-only nudge is only SAFE +// when this is true: a harness that submits but cannot report blocked leaves +// the confirm loop unable to tell an unsubmitted draft from a pending +// permission dialog, so an Enter meant to resubmit the draft could instead +// answer the dialog. confirmActive therefore requires BOTH signals before it +// will nudge. +// +// Only claude-code satisfies both halves: it installs the pre/post-tool-use +// trio that lets lifecycle correlate the approved tool's post with the dialog +// and clear blocked before the turn ends. codex maps permission-request to +// waiting_input and opts out (no tool trio → blocked could not be cleared). +// Every other harness simply does not implement this interface; it maps its +// permission signal to waiting_input via the shared deriver and gets the +// paste settle delay but no confirm loop. Adapters that later gain a +// correlatable blocked signal implement this interface to opt in; see the +// fork/archive/blocked-mappings branch for the prior 13-harness mapping set. +type ActivitySignaler interface { + EmitsSubmitActivity() bool + EmitsBlockedActivity() bool +} + // MetadataKeyAgentSessionID is the SessionRef.Metadata key that carries an // agent's native session id. It matches the json tag on // domain.SessionMetadata.AgentSessionID and the key the adapters read, so the diff --git a/backend/internal/ports/outbound.go b/backend/internal/ports/outbound.go index b4b618a2c8..0d8d87eaa8 100644 --- a/backend/internal/ports/outbound.go +++ b/backend/internal/ports/outbound.go @@ -68,7 +68,9 @@ type ClaimOutcome struct { OwnerTerminated bool } -// AgentMessenger injects a message into a running agent. +// AgentMessenger injects a message into a running agent. An empty message +// sends only the submit keystroke (Enter) — callers use it to nudge a pasted +// prompt that was not submitted; every runtime must honor this contract. type AgentMessenger interface { Send(ctx context.Context, id domain.SessionID, message string) error } diff --git a/backend/internal/ports/runtime_observations.go b/backend/internal/ports/runtime_observations.go index fe548969a7..ac2607dd3c 100644 --- a/backend/internal/ports/runtime_observations.go +++ b/backend/internal/ports/runtime_observations.go @@ -26,8 +26,18 @@ type RuntimeFacts struct { // ActivitySignal is pushed by the agent hooks. Only a Valid signal is // authoritative; a stale/absent one is ignored rather than read as idleness. +// +// Event/ToolName/ToolUseID are optional correlation facts: the AO hook +// sub-command that produced the state and, for tool-use hooks, the native +// tool call it concerns. Lifecycle uses them to clear a stale blocked state +// only when the specific approved tool finishes. A signal without an Event +// (old CLIs, adapters with no tool identity) keeps plain last-writer-wins +// state semantics. type ActivitySignal struct { Valid bool State domain.ActivityState Timestamp time.Time + Event string + ToolName string + ToolUseID string } diff --git a/backend/internal/service/session/service.go b/backend/internal/service/session/service.go index 0851f3e377..ce7dcf6bc8 100644 --- a/backend/internal/service/session/service.go +++ b/backend/internal/service/session/service.go @@ -554,6 +554,9 @@ func toAPIError(err error) error { return apierr.Conflict("SESSION_NOT_RESTORABLE", "Session is not restorable", nil) case errors.Is(err, sessionmanager.ErrTerminated): return apierr.Conflict("SESSION_TERMINATED", "Session is terminated", nil) + case errors.Is(err, sessionmanager.ErrAwaitingDecision): + return apierr.Conflict("SESSION_AWAITING_DECISION", + "Session is paused on a permission decision; answer it in the session terminal first", nil) case errors.Is(err, sessionmanager.ErrIncompleteHandle): return apierr.Conflict("SESSION_INCOMPLETE_HANDLE", "Session is missing runtime or workspace handles", nil) case errors.Is(err, sessionmanager.ErrNotResumable): diff --git a/backend/internal/service/session/service_test.go b/backend/internal/service/session/service_test.go index 9dea20257d..d87354a6b5 100644 --- a/backend/internal/service/session/service_test.go +++ b/backend/internal/service/session/service_test.go @@ -583,6 +583,7 @@ func TestToAPIErrorMapsWorkspaceBranchSentinels(t *testing.T) { {"runtime prerequisite missing", fmt.Errorf("spawn: %w: tmux required on macOS/Linux but not in PATH", ports.ErrRuntimePrerequisite), apierr.KindInvalid, "RUNTIME_PREREQUISITE_MISSING"}, {"unknown harness", fmt.Errorf("spawn: %w: %q", sessionmanager.ErrUnknownHarness, "bogus"), apierr.KindInvalid, "UNKNOWN_HARNESS"}, {"missing harness", fmt.Errorf("spawn: %w: configure project worker.agent or pass --harness", sessionmanager.ErrMissingHarness), apierr.KindInvalid, "AGENT_REQUIRED"}, + {"awaiting decision", fmt.Errorf("send mer-1: %w", sessionmanager.ErrAwaitingDecision), apierr.KindConflict, "SESSION_AWAITING_DECISION"}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { diff --git a/backend/internal/service/session/status.go b/backend/internal/service/session/status.go index 1bacbbde1e..50062eb984 100644 --- a/backend/internal/service/session/status.go +++ b/backend/internal/service/session/status.go @@ -32,7 +32,7 @@ func deriveStatus(rec domain.SessionRecord, prs []domain.PRFacts, now time.Time, return domain.StatusTerminated } - if rec.Activity.State == domain.ActivityWaitingInput { + if rec.Activity.State.NeedsInput() { return domain.StatusNeedsInput } diff --git a/backend/internal/service/session/status_test.go b/backend/internal/service/session/status_test.go index f989ed49ca..a7e3d1a420 100644 --- a/backend/internal/service/session/status_test.go +++ b/backend/internal/service/session/status_test.go @@ -42,6 +42,7 @@ func TestServiceDerivesStatusFromSessionFactsAndPR(t *testing.T) { {"terminated", statusRec(domain.ActivityExited, true), nil, false, domain.StatusTerminated}, {"merged-pr", statusRec(domain.ActivityIdle, true), statusPR(domain.PRFacts{Merged: true}), false, domain.StatusMerged}, {"needs-input", statusRec(domain.ActivityWaitingInput, false), statusPR(domain.PRFacts{CI: domain.CIFailing}), false, domain.StatusNeedsInput}, + {"needs-input-blocked", statusRec(domain.ActivityBlocked, false), statusPR(domain.PRFacts{CI: domain.CIFailing}), false, domain.StatusNeedsInput}, {"ci-failed", statusRec(domain.ActivityIdle, false), statusPR(domain.PRFacts{CI: domain.CIFailing}), false, domain.StatusCIFailed}, {"draft", statusRec(domain.ActivityIdle, false), statusPR(domain.PRFacts{Draft: true}), false, domain.StatusDraft}, {"changes-requested", statusRec(domain.ActivityIdle, false), statusPR(domain.PRFacts{Review: domain.ReviewChangesRequest}), false, domain.StatusChangesRequested}, diff --git a/backend/internal/session_manager/manager.go b/backend/internal/session_manager/manager.go index 76d3cf3c46..c8fcdb898a 100644 --- a/backend/internal/session_manager/manager.go +++ b/backend/internal/session_manager/manager.go @@ -17,6 +17,7 @@ import ( "github.com/aoagents/agent-orchestrator/backend/internal/domain" "github.com/aoagents/agent-orchestrator/backend/internal/ports" aoprocess "github.com/aoagents/agent-orchestrator/backend/internal/process" + "github.com/aoagents/agent-orchestrator/backend/internal/sessionguard" "github.com/aoagents/agent-orchestrator/backend/internal/skillassets" ) @@ -47,6 +48,12 @@ var ( // session. The API maps it to a 409 so a double-submit does not race two // teardown/relaunch cycles over one worktree. ErrSwitchInProgress = errors.New("session: switch already in progress") + // ErrAwaitingDecision means the session is paused on a pending + // permission/approval dialog. Send refuses to paste into it: the runtime + // appends Enter after every paste, and an Enter into a decision dialog + // would answer it on the user's behalf. The API maps it to a 409; the + // caller retries once the user has answered in the terminal. + ErrAwaitingDecision = errors.New("session: awaiting a user decision") ) // Env vars a spawned process reads to learn who it is. @@ -117,7 +124,12 @@ type Manager struct { agents ports.AgentResolver workspace ports.Workspace store Store - messenger ports.AgentMessenger + // messenger is a sessionguard.Guard wrapping the raw messenger, so every + // pane write is guarded (re-read state, refuse a blocked session) without + // each call site re-deriving the check. Send/confirmActive use Deliver for + // its Outcome; Spawn/Restore use the interface-level Send for + // initial-prompt delivery, where a blocked session is impossible. + messenger *sessionguard.Guard lcm lifecycleRecorder dataDir string clock func() time.Time @@ -129,8 +141,37 @@ type Manager struct { // production); its directory is prepended to spawned sessions' PATH so the // workspace hook commands resolve back to this daemon. Tests inject a stub. executable func() (string, error) - logger *slog.Logger -} + // sendConfirm bounds the best-effort post-send confirmation that the session + // actually became active (the agent accepted the prompt). New fills in the + // sendConfirm* defaults; tests in this package shrink the timings directly. + sendConfirm sendConfirmConfig + logger *slog.Logger +} + +// sendConfirmConfig bounds the best-effort activity-confirmation loop run after +// Send. AO has no delivery ack: ao send returns 200 the moment tmux send-keys +// exits 0, and for a large multiline paste the single Enter may not submit the +// prompt — so UserPromptSubmit never fires and the orchestrator cannot tell the +// worker started. confirmActive observes the durable Activity.State (written by +// the user-prompt-submit hook) and re-sends Enter until the session is active or +// the budget is exhausted. It never fails the send. +type sendConfirmConfig struct { + // pollInterval is the gap between activity reads. + pollInterval time.Duration + // attemptDeadline is how long to wait for active after each Enter. + attemptDeadline time.Duration + // maxAttempts bounds how many times Enter is (re)sent, counting the initial + // Enter from Send itself. + maxAttempts int +} + +// Production sendConfirm bounds: 3 Enters total (1 from Send + 2 re-sends), +// each given 2s to flip the session active, polled every 300ms. +const ( + sendConfirmPollInterval = 300 * time.Millisecond + sendConfirmAttemptDeadline = 2 * time.Second + sendConfirmMaxAttempts = 3 +) // Deps are the collaborators a Session Manager needs; New wires them together. type Deps struct { @@ -165,13 +206,17 @@ func New(d Deps) *Manager { agents: d.Agents, workspace: d.Workspace, store: d.Store, - messenger: d.Messenger, lcm: d.Lifecycle, dataDir: d.DataDir, clock: d.Clock, lookPath: d.LookPath, executable: d.Executable, - logger: d.Logger, + sendConfirm: sendConfirmConfig{ + pollInterval: sendConfirmPollInterval, + attemptDeadline: sendConfirmAttemptDeadline, + maxAttempts: sendConfirmMaxAttempts, + }, + logger: d.Logger, } if m.clock == nil { // UTC so spawn-stamped CreatedAt/UpdatedAt match every other session @@ -188,6 +233,9 @@ func New(d Deps) *Manager { if m.logger == nil { m.logger = slog.Default() } + // messenger is the raw d.Messenger wrapped in a Guard (needs m.logger, so it + // is built after the logger default). + m.messenger = sessionguard.New(d.Store, d.Messenger, m.logger) return m } @@ -1380,14 +1428,169 @@ func (m *Manager) applyWorkspaceProjectPreserved(ctx context.Context, rows []por } } -// Send delivers a message to a running session's agent via the messenger. +// Send delivers a message to a running session's agent through the guarded +// pane-write primitive, then best-effort confirms the agent actually accepted +// it. The guard refuses delivery into a session that is gone, terminated, or +// paused on a permission decision (pasting there could answer the dialog); +// those refusals surface as typed sentinels so the API reports why instead of +// silently dropping the message. AO has no delivery ack: the messenger returns +// nil the moment the runtime paste + Enter commands exit 0, and for a large +// multiline prompt a single Enter may not submit (claude-code leaves it as an +// unsubmitted draft). confirmActive observes the durable Activity.State +// (flipped to active by the user-prompt-submit hook) and re-sends Enter until +// the session is active or the budget is exhausted. Confirmation never fails +// the send: it only decides whether to nudge again. func (m *Manager) Send(ctx context.Context, id domain.SessionID, message string) error { - if err := m.messenger.Send(ctx, id, message); err != nil { + outcome, err := m.messenger.Deliver(ctx, id, message) + if err != nil { return fmt.Errorf("send %s: %w", id, err) } + switch outcome { + case sessionguard.SuppressedNotFound: + return fmt.Errorf("send %s: %w", id, ErrNotFound) + case sessionguard.SuppressedTerminated: + return fmt.Errorf("send %s: %w", id, ErrTerminated) + case sessionguard.SuppressedAwaitingUser: + return fmt.Errorf("send %s: %w", id, ErrAwaitingDecision) + } + // confirmActive only helps — and is only SAFE — when the harness reports + // both a prompt-submit signal (so the loop can observe active) and a + // blocked signal it can clear mid-turn (so it can tell an unsubmitted + // draft from a pending permission dialog and never Enter into the latter). + // Only claude-code and its hook-delegators (grok/continueagent/devin) + // satisfy both; every other harness opts out via EmitsBlockedActivity — + // see ports.ActivitySignaler. + rec, ok, err := m.store.GetSession(ctx, id) + if err != nil { + // Confirmation is best-effort and never fails the send (the message + // was already delivered above); log so a store error is not swallowed + // silently. + m.logger.Warn("send: confirm skipped, session lookup failed", "sessionID", id, "error", err) + return nil + } + if !ok { + return nil + } + if m.harnessNudgeSafe(rec.Harness) { + m.confirmActive(ctx, m.messenger, id) + } return nil } +// harnessNudgeSafe reports whether the session's harness is safe to nudge with +// an Enter-only re-send (see ports.ActivitySignaler): it must emit BOTH a +// prompt-submit signal (else the loop wastes its budget never observing active) +// and a blocked signal (else an Enter meant to resubmit a draft could answer a +// permission dialog the harness cannot report). +func (m *Manager) harnessNudgeSafe(harness domain.AgentHarness) bool { + agent, ok := m.agents.Agent(harness) + if !ok { + return false + } + s, ok := agent.(ports.ActivitySignaler) + return ok && s.EmitsSubmitActivity() && s.EmitsBlockedActivity() +} + +// waitOutcome is one poll round's verdict on whether confirmActive should +// nudge again. +type waitOutcome int + +const ( + // waitTimedOut: the deadline elapsed without the session going active — + // the previous Enter likely did not land, another may help. + waitTimedOut waitOutcome = iota + // waitActive: the session went active — the prompt was accepted, done. + waitActive + // waitBlocked: the session is paused on a user decision (a pending + // permission/approval dialog) — an automated Enter could answer the dialog + // on the user's behalf, so confirmation must stop and never nudge. + waitBlocked +) + +// confirmActive re-sends Enter until the session reports ActivityActive or the +// attempt budget is exhausted. The initial Send already submitted one Enter; +// each additional attempt sends Enter again (an empty message is an Enter-only +// nudge, see ports.AgentMessenger) after waiting for Activity.State to flip. It +// is best-effort: on context cancellation, store failure, or budget exhaustion +// it returns silently (the message was already delivered; the agent may yet +// pick it up). Harnesses without a user-prompt-submit hook never flip to +// active, so the loop simply times out — Send remains successful for them. +// +// Decision safety: a session observed in ActivityBlocked stops confirmation +// immediately with no nudge — an Enter into a pending permission dialog would +// answer it for the user. Sticky ActivityWaitingInput does NOT stop the loop: +// an idle-prompt session with an unsubmitted pasted draft is exactly the case +// the nudge exists for. +func (m *Manager) confirmActive(ctx context.Context, guard *sessionguard.Guard, id domain.SessionID) { + for attempt := 1; ; attempt++ { + outcome, err := m.waitForActive(ctx, id) + if err != nil || outcome == waitActive { + return + } + if outcome == waitBlocked { + m.logger.Info("send: session awaiting a decision; skipping Enter nudge", "sessionID", id, "attempt", attempt) + return + } + if attempt >= m.sendConfirm.maxAttempts { + return + } + // Timed out with budget remaining: the previous Enter did not land. + // Nudge again with an Enter-only send. Deliver re-reads state + // immediately before pasting — a permission dialog can appear in the + // gap between waitForActive's final poll and this send, and an Enter + // into it would answer the decision. This closes the TOCTOU the + // per-poll check inside waitForActive cannot cover; a store failure + // inside the guard fails closed (no Enter on an unknown state). + nudge, nudgeErr := guard.Deliver(ctx, id, "") + if nudgeErr != nil { + m.logger.Warn("send: confirm re-send failed", "sessionID", id, "attempt", attempt, "error", nudgeErr) + return + } + if nudge != sessionguard.Sent { + // Not necessarily blocked: the session may also have terminated or + // vanished since the poll — the outcome says which. + m.logger.Info("send: session unavailable before nudge; skipping Enter nudge", "sessionID", id, "attempt", attempt, "outcome", nudge.String()) + return + } + } +} + +// waitForActive polls Activity.State for up to attemptDeadline and reports +// whether another nudge could help (see waitOutcome). Blocked is checked every +// poll so a permission dialog appearing mid-wait aborts immediately instead of +// burning the deadline. A non-nil error means polling cannot continue (ctx +// cancelled, store failure, session gone). +func (m *Manager) waitForActive(ctx context.Context, id domain.SessionID) (waitOutcome, error) { + deadlineAt := m.clock().Add(m.sendConfirm.attemptDeadline) + ticker := time.NewTicker(m.sendConfirm.pollInterval) + defer ticker.Stop() + for { + rec, ok, err := m.store.GetSession(ctx, id) + if err != nil { + return waitTimedOut, err + } + if !ok { + return waitTimedOut, fmt.Errorf("session %s not found", id) + } + switch rec.Activity.State { + case domain.ActivityActive: + return waitActive, nil + case domain.ActivityBlocked: + return waitBlocked, nil + } + if !m.clock().Before(deadlineAt) { + return waitTimedOut, nil + } + // The tick select respects ctx cancellation so a request timeout + // unblocks promptly. + select { + case <-ctx.Done(): + return waitTimedOut, ctx.Err() + case <-ticker.C: + } + } +} + // CleanupSkip reports one terminal session whose workspace was preserved // rather than reclaimed, and why. type CleanupSkip struct { @@ -1868,7 +2071,26 @@ func (m *Manager) deliverAfterStartPrompt(ctx context.Context, agent ports.Agent if err := m.waitForPromptReadiness(ctx, agent, cfg, handle); err != nil { return err } - return m.messenger.Send(ctx, id, prompt) + // Call Deliver directly (not the Guard.Send wrapper, which folds a suppressed + // outcome into nil): a freshly-spawned session can terminate or hit a + // permission dialog between readiness and prompt injection, and folding that + // into success would report a spawn/restore that never delivered its prompt. + outcome, err := m.messenger.Deliver(ctx, id, prompt) + if err != nil { + return fmt.Errorf("send %s: %w", id, err) + } + switch outcome { + case sessionguard.SuppressedNotFound: + return fmt.Errorf("send %s: %w", id, ErrNotFound) + case sessionguard.SuppressedTerminated: + return fmt.Errorf("send %s: %w", id, ErrTerminated) + case sessionguard.SuppressedAwaitingUser: + return fmt.Errorf("send %s: %w", id, ErrAwaitingDecision) + case sessionguard.SuppressedUnknown: + return fmt.Errorf("send %s: pre-write session read failed", id) + default: + return nil + } } func (m *Manager) waitForPromptReadiness(ctx context.Context, agent ports.Agent, cfg ports.LaunchConfig, handle ports.RuntimeHandle) error { diff --git a/backend/internal/session_manager/manager_test.go b/backend/internal/session_manager/manager_test.go index d9facbfd15..93289f64ec 100644 --- a/backend/internal/session_manager/manager_test.go +++ b/backend/internal/session_manager/manager_test.go @@ -783,6 +783,71 @@ func TestSpawn_AfterStartPromptFailureCleansUpWorkspaceProjectRows(t *testing.T) } } +// terminatedOnReReadStore wraps fakeStore and reports the spawned session as +// terminated every time GetSession is called AFTER CreateSession, so the guard +// re-reads a terminated row and suppresses the after-start prompt delivery. +type terminatedOnReReadStore struct { + *fakeStore + spawned domain.SessionID + saw bool +} + +func (s *terminatedOnReReadStore) CreateSession(ctx context.Context, rec domain.SessionRecord) (domain.SessionRecord, error) { + out, err := s.fakeStore.CreateSession(ctx, rec) + // fakeStore assigns the "{project}-{n}" id inside CreateSession, so capture + // it from the returned record. + s.spawned = out.ID + s.saw = true + return out, err +} + +func (s *terminatedOnReReadStore) GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) { + // Once spawn created the row, surface it as terminated so the guard's + // just-in-time re-read sees a dead session and suppresses the write. + if s.saw && id == s.spawned { + rec := s.sessions[id] + rec.IsTerminated = true + rec.Activity.State = domain.ActivityExited + return rec, true, nil + } + return s.fakeStore.GetSession(ctx, id) +} + +// TestSpawn_AfterStartPromptSuppressedTerminationFailsSpawn: if a session is +// gone by the time the after-start prompt is delivered, the Guard's Deliver +// suppresses — and deliverAfterStartPrompt must surface that as an error +// (not fold it into nil / report a successful spawn with no prompt). This is +// the case the Guard.Send wrapper used to swallow (see review on #2357). +func TestSpawn_AfterStartPromptSuppressedTerminationFailsSpawn(t *testing.T) { + base := newFakeStore() + base.projects["mer"] = domain.ProjectRecord{ID: "mer", Config: testRoleAgents()} + st := &terminatedOnReReadStore{fakeStore: base} + rt := &fakeRuntime{} + ws := &fakeWorkspace{} + msg := &fakeMessenger{} // underlying messenger is fine; suppression comes from the guard's re-read + agent := &recordingAgent{} + lcm := &fakeLCM{store: base} + m := New(Deps{ + Runtime: rt, + Agents: singleAgent{agent: afterStartAgent{recordingAgent: agent}}, + Workspace: ws, + Store: st, + Messenger: msg, + Lifecycle: lcm, + LookPath: func(string) (string, error) { return "/bin/true", nil }, + }) + + _, err := m.Spawn(ctx, ports.SpawnConfig{ProjectID: "mer", Kind: domain.KindWorker, Prompt: "fix the button"}) + if err == nil { + t.Fatal("Spawn err = nil, want failure because the after-start prompt was suppressed (session terminated)") + } + // The suppressed termination must not have been folded into success: the + // prompt was never delivered. + if len(msg.msgs) != 0 { + t.Fatalf("delivered prompts = %#v, want none (delivery was suppressed)", msg.msgs) + } +} + func TestSpawn_PromptDeliveryStrategyFailureCleansUpWorkspaceProjectRows(t *testing.T) { st := newFakeStore() st.projects["mer"] = domain.ProjectRecord{ @@ -3329,3 +3394,310 @@ func TestReconcileReap_TerminatedAndDeadTmuxLeftAlone(t *testing.T) { t.Fatalf("Destroy calls = %d, want 0", rt.destroyed) } } + +// --- Send activity-confirmation tests (issue #2342) --- + +// signalingAgent is a fakeAgent that advertises BOTH a prompt-submit and a +// blocked activity signal, so Manager.Send runs confirmActive for its harness +// (see ports.ActivitySignaler). +type signalingAgent struct{ fakeAgent } + +func (signalingAgent) EmitsSubmitActivity() bool { return true } +func (signalingAgent) EmitsBlockedActivity() bool { return true } + +// submitOnlyAgent advertises a prompt-submit signal but NOT a blocked one — a +// harness like goose/opencode/agy that submits yet installs no permission hook. +// confirmActive must refuse to nudge it (it could Enter into a decision the +// harness cannot report). +type submitOnlyAgent struct{ fakeAgent } + +func (submitOnlyAgent) EmitsSubmitActivity() bool { return true } +func (submitOnlyAgent) EmitsBlockedActivity() bool { return false } + +// newSendTestManager builds a Manager wired for Send confirmation tests with +// fast (millisecond) confirmation timings so no test waits real seconds. The +// returned messenger records every Send; the store is mutable so a test can +// flip Activity.State between polls. +func newSendTestManager(t *testing.T, agent ports.Agent, messenger ports.AgentMessenger, st *fakeStore) *Manager { + t.Helper() + rt := &fakeRuntime{} + ws := &fakeWorkspace{} + lcm := &fakeLCM{store: st} + lookPath := func(string) (string, error) { return "/bin/true", nil } + m := New(Deps{ + Runtime: rt, Agents: singleAgent{agent}, Workspace: ws, Store: st, + Messenger: messenger, Lifecycle: lcm, LookPath: lookPath, + }) + // Shrink the confirmation budget so the loop runs in milliseconds, not + // seconds. m.sendConfirm is unexported; tests live in this package. + m.sendConfirm = sendConfirmConfig{ + pollInterval: time.Millisecond, + attemptDeadline: 2 * time.Millisecond, + maxAttempts: 3, + } + return m +} + +func TestSend_SkipsConfirmForHooklessHarness(t *testing.T) { + // A harness whose adapter does NOT implement ActivitySignaler (plain + // fakeAgent) must skip confirmActive entirely: one Send, no nudges, and the + // call returns immediately without polling. + st := newFakeStore() + st.sessions["s1"] = domain.SessionRecord{ID: "s1", Harness: "claude-code"} + msg := &fakeMessenger{} + m := newSendTestManager(t, fakeAgent{}, msg, st) + + start := time.Now() + if err := m.Send(context.Background(), "s1", "hello"); err != nil { + t.Fatalf("Send: %v", err) + } + if len(msg.msgs) != 1 { + t.Fatalf("Send calls = %d, want 1 (no nudges for a hookless harness)", len(msg.msgs)) + } + // Hookless path returns within milliseconds (no 2s+ confirmation wait). + if dt := time.Since(start); dt > 250*time.Millisecond { + t.Fatalf("Send took %s for a hookless harness; confirmActive should have been skipped", dt) + } +} + +func TestSend_ConfirmsAndNudgesUntilActive(t *testing.T) { + // A signaling harness starts idle. The first nudge (Enter-only Send) should + // flip the session active, after which confirmActive stops. Net: the + // initial message plus exactly one nudge. + st := newFakeStore() + st.sessions["s1"] = domain.SessionRecord{ID: "s1", Harness: "claude-code", + Activity: domain.Activity{State: domain.ActivityIdle}} + // A messenger that flips the session active on the first Enter-only nudge, + // mimicking the agent accepting the prompt. + msg := &flipOnNudgeMessenger{sessionID: "s1", store: st} + m := newSendTestManager(t, signalingAgent{}, msg, st) + + if err := m.Send(context.Background(), "s1", "do the thing"); err != nil { + t.Fatalf("Send: %v", err) + } + if len(msg.msgs) != 2 { + t.Fatalf("Send calls = %d, want 2 (initial + one nudge)", len(msg.msgs)) + } + if msg.msgs[0] != "do the thing" { + t.Fatalf("first msg = %q, want the prompt", msg.msgs[0]) + } + if msg.msgs[1] != "" { + t.Fatalf("nudge msg = %q, want empty (Enter-only)", msg.msgs[1]) + } + if got := st.sessions["s1"].Activity.State; got != domain.ActivityActive { + t.Fatalf("Activity.State = %q, want active", got) + } +} + +func TestSend_ConfirmBudgetCapsRetries(t *testing.T) { + // A signaling harness that never goes active must still terminate: at most + // maxAttempts Sends (initial + maxAttempts-1 nudges), and Send never errors. + st := newFakeStore() + st.sessions["s1"] = domain.SessionRecord{ID: "s1", Harness: "claude-code", + Activity: domain.Activity{State: domain.ActivityIdle}} + msg := &fakeMessenger{} + m := newSendTestManager(t, signalingAgent{}, msg, st) + + if err := m.Send(context.Background(), "s1", "stuck prompt"); err != nil { + t.Fatalf("Send: %v", err) + } + if len(msg.msgs) > m.sendConfirm.maxAttempts { + t.Fatalf("Send calls = %d, want <= %d (budget cap)", len(msg.msgs), m.sendConfirm.maxAttempts) + } + if got := st.sessions["s1"].Activity.State; got == domain.ActivityActive { + t.Fatalf("Activity.State = active, want unchanged (session never went active)") + } +} + +func TestSend_BlockedSessionRejectsDelivery(t *testing.T) { + // A session paused on a permission decision (blocked) must not receive the + // paste at all: the runtime appends Enter, which could answer the dialog. + // Send surfaces ErrAwaitingDecision (the API's 409) and the messenger is + // never called, so nothing — message or nudge — reaches the pane. + st := newFakeStore() + st.sessions["s1"] = domain.SessionRecord{ID: "s1", Harness: "claude-code", + Activity: domain.Activity{State: domain.ActivityBlocked}} + msg := &fakeMessenger{} + m := newSendTestManager(t, signalingAgent{}, msg, st) + + err := m.Send(context.Background(), "s1", "status update please") + if !errors.Is(err, ErrAwaitingDecision) { + t.Fatalf("Send error = %v, want ErrAwaitingDecision", err) + } + if len(msg.msgs) != 0 { + t.Fatalf("Send calls = %d, want 0 (no paste into a pending decision)", len(msg.msgs)) + } +} + +func TestSend_NoNudgeWhenBlockedAppearsMidWait(t *testing.T) { + // The permission dialog can appear between polls (e.g. the delivered prompt + // itself triggered a tool approval). The confirm loop must abort on the + // first blocked observation instead of nudging after the deadline. + st := newFakeStore() + st.sessions["s1"] = domain.SessionRecord{ID: "s1", Harness: "claude-code", + Activity: domain.Activity{State: domain.ActivityIdle}} + msg := &blockOnSendMessenger{sessionID: "s1", store: st} + m := newSendTestManager(t, signalingAgent{}, msg, st) + + if err := m.Send(context.Background(), "s1", "run the migration"); err != nil { + t.Fatalf("Send: %v", err) + } + if len(msg.msgs) != 1 { + t.Fatalf("Send calls = %d, want 1 (blocked observed mid-confirm, no nudge)", len(msg.msgs)) + } +} + +func TestSend_StillNudgesWhenWaitingInput(t *testing.T) { + // waiting_input (an idle prompt awaiting the next instruction) is the + // PRIMARY nudge scenario: a long-idle worker with an unsubmitted pasted + // draft. The decision-safety guard must not disable it. + st := newFakeStore() + st.sessions["s1"] = domain.SessionRecord{ID: "s1", Harness: "claude-code", + Activity: domain.Activity{State: domain.ActivityWaitingInput}} + msg := &flipOnNudgeMessenger{sessionID: "s1", store: st} + m := newSendTestManager(t, signalingAgent{}, msg, st) + + if err := m.Send(context.Background(), "s1", "do the thing"); err != nil { + t.Fatalf("Send: %v", err) + } + if len(msg.msgs) != 2 { + t.Fatalf("Send calls = %d, want 2 (initial + one nudge for waiting_input)", len(msg.msgs)) + } + if msg.msgs[1] != "" { + t.Fatalf("nudge msg = %q, want empty (Enter-only)", msg.msgs[1]) + } +} + +// blockOnSendMessenger records sends and flips the session to ActivityBlocked +// right after the initial message is delivered, simulating a prompt that +// immediately triggers a tool-permission dialog. +type blockOnSendMessenger struct { + msgs []string + sessionID domain.SessionID + store *fakeStore +} + +func (m *blockOnSendMessenger) Send(_ context.Context, _ domain.SessionID, msg string) error { + m.msgs = append(m.msgs, msg) + if rec, ok := m.store.sessions[m.sessionID]; ok { + rec.Activity.State = domain.ActivityBlocked + m.store.sessions[m.sessionID] = rec + } + return nil +} + +func TestSend_NoNudgeWhenBlockedAppearsBeforeNudge(t *testing.T) { + // The TOCTOU the per-poll check cannot cover: the session is not blocked on + // waitForActive's final poll, but a permission dialog lands in the gap + // before the Enter-only nudge. The just-in-time re-read in confirmActive + // must catch it — exactly one Send, no nudge. + st := newFakeStore() + st.sessions["s1"] = domain.SessionRecord{ID: "s1", Harness: "claude-code", + Activity: domain.Activity{State: domain.ActivityIdle}} + // blockAfterFirstReadStore flips the session to blocked on read #4. The + // deterministic read sequence (attemptDeadline 0 makes waitForActive do + // exactly one poll): #1 Deliver's pre-paste read, #2 Send's harness lookup, + // #3 waitForActive's poll (idle → timeout), #4 the JIT pre-nudge re-read — + // which is the first to see blocked, landing the flip in the exact + // post-final-poll / pre-nudge window this test exists to cover. + bst := &blockAfterFirstReadStore{fakeStore: st, id: "s1"} + msg := &fakeMessenger{} + m := New(Deps{ + Runtime: &fakeRuntime{}, Agents: singleAgent{signalingAgent{}}, Workspace: &fakeWorkspace{}, + Store: bst, Messenger: msg, Lifecycle: &fakeLCM{store: st}, + LookPath: func(string) (string, error) { return "/bin/true", nil }, + }) + m.sendConfirm = sendConfirmConfig{pollInterval: time.Millisecond, attemptDeadline: 0, maxAttempts: 3} + + if err := m.Send(context.Background(), "s1", "run the migration"); err != nil { + t.Fatalf("Send: %v", err) + } + if len(msg.msgs) != 1 { + t.Fatalf("Send calls = %d, want 1 (blocked appeared before nudge, JIT re-read caught it)", len(msg.msgs)) + } + if bst.reads < 4 { + t.Fatalf("GetSession reads = %d, want >= 4 (the JIT pre-nudge re-read must have run)", bst.reads) + } +} + +func TestSend_SkipsConfirmForSubmitOnlyHarness(t *testing.T) { + // A harness that submits but cannot report blocked (goose/opencode/agy) is + // NOT nudge-safe: confirmActive must be skipped entirely, so an Enter can + // never reach a permission dialog the harness could not have signalled. + st := newFakeStore() + st.sessions["s1"] = domain.SessionRecord{ID: "s1", Harness: "goose", + Activity: domain.Activity{State: domain.ActivityIdle}} + msg := &fakeMessenger{} + m := newSendTestManager(t, submitOnlyAgent{}, msg, st) + + if err := m.Send(context.Background(), "s1", "do the thing"); err != nil { + t.Fatalf("Send: %v", err) + } + if len(msg.msgs) != 1 { + t.Fatalf("Send calls = %d, want 1 (submit-only harness must not be nudged)", len(msg.msgs)) + } +} + +func TestHarnessNudgeSafe(t *testing.T) { + m := New(Deps{Agents: singleAgent{agent: fakeAgent{}}}) + if m.harnessNudgeSafe("claude-code") { + t.Fatalf("hookless agent reported as nudge-safe") + } + m2 := New(Deps{Agents: singleAgent{agent: signalingAgent{}}}) + if !m2.harnessNudgeSafe("claude-code") { + t.Fatalf("submit+blocked agent not reported as nudge-safe") + } + m3 := New(Deps{Agents: singleAgent{agent: submitOnlyAgent{}}}) + if m3.harnessNudgeSafe("claude-code") { + t.Fatalf("submit-only agent (no blocked signal) reported as nudge-safe") + } + m4 := New(Deps{Agents: missingAgents{}}) + if m4.harnessNudgeSafe("claude-code") { + t.Fatalf("unresolved harness reported as nudge-safe") + } +} + +// blockAfterFirstReadStore wraps fakeStore and flips the session to +// ActivityBlocked on the FOURTH GetSession call, so with attemptDeadline 0 the +// first read to observe blocked is confirmActive's just-in-time pre-nudge +// re-read (reads #1-#3 are Deliver's pre-paste read, Send's harness lookup, +// and waitForActive's single poll — see TestSend_NoNudgeWhenBlockedAppearsBeforeNudge). +type blockAfterFirstReadStore struct { + *fakeStore + id domain.SessionID + reads int +} + +func (s *blockAfterFirstReadStore) GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) { + s.reads++ + if s.reads >= 4 { + if rec, ok := s.sessions[s.id]; ok { + rec.Activity.State = domain.ActivityBlocked + s.sessions[s.id] = rec + } + } + return s.fakeStore.GetSession(ctx, id) +} + +// flipOnNudgeMessenger records sends like fakeMessenger and additionally flips a +// session to ActivityActive the first time it receives an Enter-only nudge (an +// empty message), simulating the agent accepting the prompt after the retry. +type flipOnNudgeMessenger struct { + msgs []string + sessionID domain.SessionID + store *fakeStore + flipped bool +} + +func (m *flipOnNudgeMessenger) Send(_ context.Context, _ domain.SessionID, msg string) error { + m.msgs = append(m.msgs, msg) + if msg == "" && !m.flipped { + rec, ok := m.store.sessions[m.sessionID] + if ok { + rec.Activity.State = domain.ActivityActive + m.store.sessions[m.sessionID] = rec + } + m.flipped = true + } + return nil +} diff --git a/backend/internal/sessionguard/guard.go b/backend/internal/sessionguard/guard.go new file mode 100644 index 0000000000..3e159ed02b --- /dev/null +++ b/backend/internal/sessionguard/guard.go @@ -0,0 +1,156 @@ +// Package sessionguard owns the one invariant every write into a live +// session's pane must satisfy: re-read the session immediately before writing +// and refuse when the paste could land somewhere only the user may act. The +// runtime appends Enter after every paste, so a write into a session paused on +// a permission/approval dialog would answer the decision on the user's behalf +// — an unrecoverable action, unlike a skipped message which callers re-attempt +// or surface. Every pane-writing path (user sends, post-send Enter nudges, +// lifecycle reaction nudges) funnels through this guard so the stale-state +// check lives in one tested place instead of being re-derived per call-site. +package sessionguard + +import ( + "context" + "fmt" + "log/slog" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" + "github.com/aoagents/agent-orchestrator/backend/internal/ports" +) + +// SessionReader is the single store read the guard needs: the session's +// current liveness and activity state. +type SessionReader interface { + GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) +} + +// Outcome reports what a guarded write did. Anything other than Sent means the +// message did NOT reach the pane; callers that record delivery must not stamp +// a suppressed write as delivered. +type Outcome int + +const ( + // SuppressedUnknown is returned when the pre-write session read failed, so + // the state is unknown and the guard failed closed. Deliberately the zero + // value — a forgotten assignment must never read as a successful send. + SuppressedUnknown Outcome = iota + // Sent means the message was written to the session's pane (a messenger + // failure surfaces as Sent plus a non-nil error: the write was attempted). + Sent + // SuppressedNotFound means no session row exists for the id. + SuppressedNotFound + // SuppressedTerminated means the session is terminated; its pane is gone + // or about to be reaped. + SuppressedTerminated + // SuppressedAwaitingUser means the session awaits the human — blocked on a + // permission decision (Deliver and Nudge), or waiting at the prompt for + // the next instruction (Nudge only). + SuppressedAwaitingUser +) + +// String names the outcome for logs. +func (o Outcome) String() string { + switch o { + case Sent: + return "sent" + case SuppressedNotFound: + return "suppressed_not_found" + case SuppressedTerminated: + return "suppressed_terminated" + case SuppressedAwaitingUser: + return "suppressed_awaiting_user" + default: + return "suppressed_unknown" + } +} + +// Guard is the guarded pane-write primitive shared by the session manager and +// lifecycle. It takes no locks of its own, so callers may hold theirs across a +// call (lifecycle's sendOnce calls it under react.mu). It implements +// ports.AgentMessenger (via Send) so it can transparently replace a raw +// messenger wherever only the error matters. +type Guard struct { + store SessionReader + messenger ports.AgentMessenger + logger *slog.Logger +} + +var _ ports.AgentMessenger = (*Guard)(nil) + +// New builds a Guard over the store it re-reads and the messenger it writes +// through. A nil logger falls back to slog.Default(). +func New(store SessionReader, messenger ports.AgentMessenger, logger *slog.Logger) *Guard { + if logger == nil { + logger = slog.Default() + } + return &Guard{store: store, messenger: messenger, logger: logger} +} + +// Send satisfies ports.AgentMessenger so a Guard can sit in for the raw +// messenger. It applies the Deliver policy but FOLDS a suppressed outcome into +// nil: a caller that learns only "did Send error?" cannot tell that the write +// was actually refused. That is fine for callers that only need a best-effort +// delivery, but paths whose success CONTRACT depends on the write landing +// (after-start prompt delivery in Spawn/Restore) must call Deliver directly and +// map non-Sent outcomes to an error, or a session that terminates or blocks +// before injection is reported as a successful spawn with a prompt that was +// never delivered. +func (g *Guard) Send(ctx context.Context, id domain.SessionID, msg string) error { + _, err := g.Deliver(ctx, id, msg) + return err +} + +// Deliver writes a user-initiated message (or its Enter-only re-submit: an +// empty msg) into the session. It refuses only when the session is blocked on +// a pending decision — waiting_input does NOT suppress, because an agent +// sitting at an idle prompt is exactly where a user message (or the Enter that +// submits its unsent draft) belongs. +func (g *Guard) Deliver(ctx context.Context, id domain.SessionID, msg string) (Outcome, error) { + return g.send(ctx, id, msg, func(state domain.ActivityState) bool { + return state == domain.ActivityBlocked + }) +} + +// Nudge writes an AO-initiated (unsolicited) message into the session. It +// refuses whenever the session awaits the human in any form — blocked on a +// decision or waiting at the prompt — because an automated paste+Enter there +// either answers a dialog or submits text the user never saw. +func (g *Guard) Nudge(ctx context.Context, id domain.SessionID, msg string) (Outcome, error) { + return g.send(ctx, id, msg, func(state domain.ActivityState) bool { + return state.NeedsInput() + }) +} + +// send re-reads the session immediately before pasting so the window between +// "state looked safe" and "bytes hit the pane" is as small as this process can +// make it. It is not atomic against the agent itself — a dialog can still +// appear mid-paste — but the just-in-time read is the strongest guarantee +// available without scraping the terminal. Fail closed: a store error +// suppresses the write rather than pressing Enter on an unknown state. +func (g *Guard) send(ctx context.Context, id domain.SessionID, msg string, refuse func(domain.ActivityState) bool) (Outcome, error) { + rec, ok, err := g.store.GetSession(ctx, id) + if err != nil { + return SuppressedUnknown, fmt.Errorf("guard %s: read session: %w", id, err) + } + if !ok { + g.logger.Info("sessionguard: write suppressed", "sessionID", id, "reason", "not_found") + return SuppressedNotFound, nil + } + // ActivityExited is refused alongside IsTerminated as defense-in-depth: + // every exited writer today also sets IsTerminated, but a pane whose agent + // exited execs an interactive shell, so a paste+Enter there would run the + // message as shell commands — the invariant must not depend on writer + // discipline alone. + if rec.IsTerminated || rec.Activity.State == domain.ActivityExited { + g.logger.Info("sessionguard: write suppressed", "sessionID", id, "reason", "terminated") + return SuppressedTerminated, nil + } + if refuse(rec.Activity.State) { + g.logger.Info("sessionguard: write suppressed", "sessionID", id, "reason", "awaiting_user", "state", string(rec.Activity.State)) + return SuppressedAwaitingUser, nil + } + if err := g.messenger.Send(ctx, id, msg); err != nil { + return Sent, fmt.Errorf("guard %s: send: %w", id, err) + } + return Sent, nil +} diff --git a/backend/internal/sessionguard/guard_test.go b/backend/internal/sessionguard/guard_test.go new file mode 100644 index 0000000000..a82beea9bc --- /dev/null +++ b/backend/internal/sessionguard/guard_test.go @@ -0,0 +1,112 @@ +package sessionguard + +import ( + "context" + "errors" + "testing" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" +) + +type fakeStore struct { + rec domain.SessionRecord + ok bool + err error +} + +func (s *fakeStore) GetSession(_ context.Context, _ domain.SessionID) (domain.SessionRecord, bool, error) { + return s.rec, s.ok, s.err +} + +type fakeMessenger struct { + sent []string + err error +} + +func (m *fakeMessenger) Send(_ context.Context, _ domain.SessionID, msg string) error { + m.sent = append(m.sent, msg) + return m.err +} + +func record(state domain.ActivityState, terminated bool) domain.SessionRecord { + return domain.SessionRecord{ID: "s1", IsTerminated: terminated, Activity: domain.Activity{State: state}} +} + +func TestGuard_OutcomeByState(t *testing.T) { + cases := []struct { + name string + rec domain.SessionRecord + ok bool + wantDeliver Outcome + wantNudge Outcome + }{ + {"active", record(domain.ActivityActive, false), true, Sent, Sent}, + {"idle", record(domain.ActivityIdle, false), true, Sent, Sent}, + // waiting_input is the split that motivates two methods: a user message + // (or its Enter re-submit) belongs at an idle prompt; an unsolicited + // automated nudge does not. + {"waiting_input", record(domain.ActivityWaitingInput, false), true, Sent, SuppressedAwaitingUser}, + {"blocked", record(domain.ActivityBlocked, false), true, SuppressedAwaitingUser, SuppressedAwaitingUser}, + // exited is refused even without IsTerminated: the pane holds an + // interactive shell after agent exit, so a paste would execute there. + {"exited", record(domain.ActivityExited, false), true, SuppressedTerminated, SuppressedTerminated}, + {"terminated", record(domain.ActivityIdle, true), true, SuppressedTerminated, SuppressedTerminated}, + {"missing", domain.SessionRecord{}, false, SuppressedNotFound, SuppressedNotFound}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + for method, want := range map[string]Outcome{"Deliver": tc.wantDeliver, "Nudge": tc.wantNudge} { + msgr := &fakeMessenger{} + g := New(&fakeStore{rec: tc.rec, ok: tc.ok}, msgr, nil) + var got Outcome + var err error + if method == "Deliver" { + got, err = g.Deliver(context.Background(), "s1", "hello") + } else { + got, err = g.Nudge(context.Background(), "s1", "hello") + } + if err != nil { + t.Fatalf("%s: unexpected error: %v", method, err) + } + if got != want { + t.Errorf("%s: outcome = %v, want %v", method, got, want) + } + if wantSent := want == Sent; (len(msgr.sent) == 1) != wantSent { + t.Errorf("%s: messenger sends = %d, want sent=%v", method, len(msgr.sent), wantSent) + } + } + }) + } +} + +func TestGuard_StoreErrorFailsClosed(t *testing.T) { + msgr := &fakeMessenger{} + g := New(&fakeStore{err: errors.New("db locked")}, msgr, nil) + for name, call := range map[string]func() (Outcome, error){ + "Deliver": func() (Outcome, error) { return g.Deliver(context.Background(), "s1", "x") }, + "Nudge": func() (Outcome, error) { return g.Nudge(context.Background(), "s1", "x") }, + } { + got, err := call() + if err == nil { + t.Fatalf("%s: want error from store failure", name) + } + if got != SuppressedUnknown { + t.Errorf("%s: outcome = %v, want SuppressedUnknown", name, got) + } + } + if len(msgr.sent) != 0 { + t.Errorf("messenger was called %d times on unknown state, want 0", len(msgr.sent)) + } +} + +func TestGuard_MessengerErrorIsSentPlusError(t *testing.T) { + sendErr := errors.New("pane gone") + g := New(&fakeStore{rec: record(domain.ActivityActive, false), ok: true}, &fakeMessenger{err: sendErr}, nil) + got, err := g.Deliver(context.Background(), "s1", "x") + if !errors.Is(err, sendErr) { + t.Fatalf("error = %v, want wrapped %v", err, sendErr) + } + if got != Sent { + t.Errorf("outcome = %v, want Sent (the write was attempted)", got) + } +} diff --git a/docs/architecture.md b/docs/architecture.md index f35aab7552..91fb1ffe1f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -35,7 +35,7 @@ flowchart LR The only persistent session state is: -- `activity_state` — What the agent last reported (`active`, `idle`, `waiting_input`, `exited`) +- `activity_state` — What the agent last reported (`active`, `idle`, `waiting_input`, `blocked`, `exited`). `waiting_input` is an agent at an empty prompt awaiting its next instruction; `blocked` is an agent stopped on a pending permission/approval decision — automation must never inject input into a blocked session. - `is_terminated` — Whether the session should be treated as over - PR facts — `pr`, `pr_checks`, `pr_comment` tables @@ -427,7 +427,7 @@ The `service.Session` computes display status from durable facts using this prec flowchart TD CheckTerm{is_terminated?} CheckTerm -->|Yes| PRMerged{PR merged?} - CheckTerm -->|No| CheckWait{activity_state
== waiting_input?} + CheckTerm -->|No| CheckWait{activity_state in
waiting_input, blocked?} PRMerged -->|Yes| Merged[merged] PRMerged -->|No| Terminated[terminated] @@ -526,7 +526,7 @@ stateDiagram-v2 Spawning --> Active: MarkSpawned Active --> Idle: activity_state = idle Active --> Working: activity_state = active - Active --> Waiting: activity_state = waiting_input + Active --> Waiting: activity_state = waiting_input / blocked Active --> Exited: activity_state = exited Working --> Active: work completes Waiting --> Active: user responds diff --git a/frontend/src/api/schema.ts b/frontend/src/api/schema.ts index 6f78883725..c3f875897e 100644 --- a/frontend/src/api/schema.ts +++ b/frontend/src/api/schema.ts @@ -1026,11 +1026,17 @@ export interface components { session: components["schemas"]["ControllersSessionView"]; }; SetActivityRequest: { + /** @description AO hook sub-command that produced this state (e.g. post-tool-use). */ + event?: string; /** * @description Agent activity state reported by an agent hook. * @enum {string} */ - state: "active" | "idle" | "waiting_input" | "exited"; + state: "active" | "idle" | "waiting_input" | "blocked" | "exited"; + /** @description Native tool name, for tool-use hook events. */ + toolName?: string; + /** @description Native tool-use id, for tool-use hook events. */ + toolUseId?: string; }; SetActivityResponse: { ok: boolean; @@ -3196,6 +3202,15 @@ export interface operations { "application/json": components["schemas"]["APIError"]; }; }; + /** @description Conflict */ + 409: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; /** @description Internal Server Error */ 500: { headers: { diff --git a/frontend/src/renderer/components/SessionInspector.tsx b/frontend/src/renderer/components/SessionInspector.tsx index 56067507ab..34a16862e8 100644 --- a/frontend/src/renderer/components/SessionInspector.tsx +++ b/frontend/src/renderer/components/SessionInspector.tsx @@ -388,6 +388,7 @@ const ACTIVITY_PILL: Record { }); }); +describe("toSessionActivity", () => { + it.each(["active", "idle", "waiting_input", "blocked", "exited"] as const)( + "passes through the known state %s", + (state) => { + expect(toSessionActivity({ state })?.state).toBe(state); + }, + ); + + it("falls back to unknown for an unrecognized state", () => { + expect(toSessionActivity({ state: "bogus" })?.state).toBe("unknown"); + }); + + it("returns undefined for a missing activity", () => { + expect(toSessionActivity(undefined)).toBeUndefined(); + expect(toSessionActivity(null)).toBeUndefined(); + }); +}); + describe("workerDisplayStatus", () => { it("prefers an explicit displayStatus override", () => { expect(workerDisplayStatus(sessionWith({ status: "ci_failed", displayStatus: "done" }))).toBe("done"); diff --git a/frontend/src/renderer/types/workspace.ts b/frontend/src/renderer/types/workspace.ts index fc8ce08b8b..7675ea631a 100644 --- a/frontend/src/renderer/types/workspace.ts +++ b/frontend/src/renderer/types/workspace.ts @@ -35,9 +35,9 @@ export function toSessionStatus(status?: string, isTerminated = false): SessionS return isTerminated ? "terminated" : "unknown"; } -export type SessionActivityState = "active" | "idle" | "waiting_input" | "exited" | "unknown"; +export type SessionActivityState = "active" | "idle" | "waiting_input" | "blocked" | "exited" | "unknown"; -const sessionActivityStates = new Set(["active", "idle", "waiting_input", "exited"]); +const sessionActivityStates = new Set(["active", "idle", "waiting_input", "blocked", "exited"]); export type SessionActivity = { state: SessionActivityState;