diff --git a/backend/internal/adapters/agent/activitydispatch/dispatch.go b/backend/internal/adapters/agent/activitydispatch/dispatch.go new file mode 100644 index 000000000..b163ec87f --- /dev/null +++ b/backend/internal/adapters/agent/activitydispatch/dispatch.go @@ -0,0 +1,32 @@ +// Package activitydispatch is the single source of truth mapping the agent +// token in `ao hooks ` onto the function that interprets that +// agent's hook callbacks as an AO activity state. +package activitydispatch + +import ( + "github.com/aoagents/agent-orchestrator/backend/internal/adapters/agent/claudecode" + "github.com/aoagents/agent-orchestrator/backend/internal/adapters/agent/codex" + "github.com/aoagents/agent-orchestrator/backend/internal/adapters/agent/opencode" + "github.com/aoagents/agent-orchestrator/backend/internal/domain" +) + +// DeriveFunc maps a native agent hook event and its raw stdin payload onto an AO +// activity state. ok=false means the event carries no activity signal. +type DeriveFunc func(event string, payload []byte) (domain.ActivityState, bool) + +// Derivers maps the agent token in `ao hooks ` to its deriver. +var Derivers = map[string]DeriveFunc{ + "claude-code": claudecode.DeriveActivityState, + "codex": codex.DeriveActivityState, + "opencode": opencode.DeriveActivityState, +} + +// Derive looks up the deriver for an agent token and applies it. ok=false when +// the token has no registered deriver or the event carries no activity signal. +func Derive(agent, event string, payload []byte) (domain.ActivityState, bool) { + derive, found := Derivers[agent] + if !found { + return "", false + } + return derive(event, payload) +} diff --git a/backend/internal/adapters/agent/claudecode/activity.go b/backend/internal/adapters/agent/claudecode/activity.go new file mode 100644 index 000000000..a4b24002a --- /dev/null +++ b/backend/internal/adapters/agent/claudecode/activity.go @@ -0,0 +1,51 @@ +package claudecode + +import ( + "encoding/json" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" +) + +// DeriveActivityState maps a Claude Code hook event and its native stdin payload +// onto an AO activity state. The bool is false when the event carries no +// activity signal. +func DeriveActivityState(event string, payload []byte) (domain.ActivityState, bool) { + switch event { + case "user-prompt-submit": + return domain.ActivityActive, true + case "stop": + return domain.ActivityIdle, true + case "notification": + return notificationState(payload) + case "session-end": + return sessionEndState(payload) + default: + return "", false + } +} + +func notificationState(payload []byte) (domain.ActivityState, bool) { + var p struct { + NotificationType string `json:"notification_type"` + } + _ = json.Unmarshal(payload, &p) + switch p.NotificationType { + case "idle_prompt", "permission_prompt": + return domain.ActivityWaitingInput, true + default: + return "", false + } +} + +func sessionEndState(payload []byte) (domain.ActivityState, bool) { + var p struct { + Reason string `json:"reason"` + } + _ = json.Unmarshal(payload, &p) + switch p.Reason { + case "clear", "resume": + return "", false + default: + return domain.ActivityExited, true + } +} diff --git a/backend/internal/adapters/agent/claudecode/activity_test.go b/backend/internal/adapters/agent/claudecode/activity_test.go new file mode 100644 index 000000000..bf25a0519 --- /dev/null +++ b/backend/internal/adapters/agent/claudecode/activity_test.go @@ -0,0 +1,40 @@ +package claudecode + +import ( + "testing" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" +) + +func TestDeriveActivityState(t *testing.T) { + tests := []struct { + name string + event string + payload string + want domain.ActivityState + wantOK bool + }{ + {"user prompt -> active", "user-prompt-submit", `{}`, domain.ActivityActive, true}, + {"stop -> idle", "stop", `{}`, domain.ActivityIdle, true}, + {"notification idle_prompt -> waiting_input", "notification", `{"notification_type":"idle_prompt"}`, domain.ActivityWaitingInput, true}, + {"notification permission_prompt -> waiting_input", "notification", `{"notification_type":"permission_prompt"}`, domain.ActivityWaitingInput, true}, + {"notification auth_success -> no signal", "notification", `{"notification_type":"auth_success"}`, "", false}, + {"notification malformed payload -> no signal", "notification", `not json`, "", false}, + {"session-end logout -> exited", "session-end", `{"reason":"logout"}`, domain.ActivityExited, true}, + {"session-end prompt_input_exit -> exited", "session-end", `{"reason":"prompt_input_exit"}`, domain.ActivityExited, true}, + {"session-end absent reason -> exited", "session-end", `{}`, domain.ActivityExited, true}, + {"session-end clear -> no signal", "session-end", `{"reason":"clear"}`, "", false}, + {"session-end resume -> no signal", "session-end", `{"reason":"resume"}`, "", false}, + {"session-start -> no signal", "session-start", `{}`, "", false}, + {"unknown event -> no signal", "frobnicate", `{}`, "", false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := DeriveActivityState(tt.event, []byte(tt.payload)) + if got != tt.want || ok != tt.wantOK { + t.Fatalf("DeriveActivityState(%q, %q) = (%q, %v), want (%q, %v)", + tt.event, tt.payload, got, ok, tt.want, tt.wantOK) + } + }) + } +} diff --git a/backend/internal/adapters/agent/claudecode/claudecode_test.go b/backend/internal/adapters/agent/claudecode/claudecode_test.go index 6fea03a9a..c512d79f1 100644 --- a/backend/internal/adapters/agent/claudecode/claudecode_test.go +++ b/backend/internal/adapters/agent/claudecode/claudecode_test.go @@ -214,6 +214,14 @@ func TestGetAgentHooksInstallsClaudeHooks(t *testing.T) { if m := matcherForCommand(config.Hooks["UserPromptSubmit"], "ao hooks claude-code user-prompt-submit"); m != nil { t.Fatalf("UserPromptSubmit matcher = %v, want none", m) } + // Notification and SessionEnd install with no matcher; the handler filters + // on the payload. + if m := matcherForCommand(config.Hooks["Notification"], "ao hooks claude-code notification"); m != nil { + t.Fatalf("Notification matcher = %v, want none", m) + } + if m := matcherForCommand(config.Hooks["SessionEnd"], "ao hooks claude-code session-end"); m != nil { + t.Fatalf("SessionEnd matcher = %v, want none", m) + } } func TestUninstallHooksRemovesClaudeHooks(t *testing.T) { diff --git a/backend/internal/adapters/agent/claudecode/hooks.go b/backend/internal/adapters/agent/claudecode/hooks.go index 56c45c1f0..8f43aca6a 100644 --- a/backend/internal/adapters/agent/claudecode/hooks.go +++ b/backend/internal/adapters/agent/claudecode/hooks.go @@ -51,20 +51,27 @@ type claudeHookSpec struct { var claudeStartupMatcher = "startup" // claudeManagedHooks is the source of truth for the hooks AO installs: -// SessionStart (under the "startup" matcher), UserPromptSubmit, and Stop. Each -// reports normalized session metadata back into AO's store. +// SessionStart (under the "startup" matcher), UserPromptSubmit, 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. var claudeManagedHooks = []claudeHookSpec{ {Event: "SessionStart", Matcher: &claudeStartupMatcher, Command: claudeHookCommandPrefix + "session-start"}, {Event: "UserPromptSubmit", Command: claudeHookCommandPrefix + "user-prompt-submit"}, {Event: "Stop", Command: claudeHookCommandPrefix + "stop"}, + {Event: "Notification", Command: claudeHookCommandPrefix + "notification"}, + {Event: "SessionEnd", Command: claudeHookCommandPrefix + "session-end"}, } // GetAgentHooks installs AO's Claude Code hooks into the worktree-local // .claude/settings.local.json file (the per-session local settings, not the // shared .claude/settings.json). The hooks (SessionStart, UserPromptSubmit, -// Stop) report normalized session metadata back into AO's store. Existing -// hooks and unrelated settings are preserved, and duplicate AO commands -// are not appended, so the install is idempotent. +// Stop, Notification, SessionEnd) report normalized session metadata and +// activity-state signals back into AO's store. Existing hooks and unrelated +// settings are preserved, and duplicate AO commands are not appended, so +// the install is idempotent. func (p *Plugin) GetAgentHooks(ctx context.Context, cfg ports.WorkspaceHookConfig) error { if err := ctx.Err(); err != nil { return err diff --git a/backend/internal/adapters/agent/codex/activity.go b/backend/internal/adapters/agent/codex/activity.go new file mode 100644 index 000000000..6b08cb30d --- /dev/null +++ b/backend/internal/adapters/agent/codex/activity.go @@ -0,0 +1,18 @@ +package codex + +import "github.com/aoagents/agent-orchestrator/backend/internal/domain" + +// DeriveActivityState maps a Codex hook event onto an AO activity state. The +// bool is false when the event carries no activity signal. +func DeriveActivityState(event string, _ []byte) (domain.ActivityState, bool) { + switch event { + case "user-prompt-submit": + return domain.ActivityActive, true + case "permission-request": + return domain.ActivityWaitingInput, true + case "stop": + return domain.ActivityIdle, true + default: + return "", false + } +} diff --git a/backend/internal/adapters/agent/codex/activity_test.go b/backend/internal/adapters/agent/codex/activity_test.go new file mode 100644 index 000000000..bf120f9cc --- /dev/null +++ b/backend/internal/adapters/agent/codex/activity_test.go @@ -0,0 +1,31 @@ +package codex + +import ( + "testing" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" +) + +func TestDeriveActivityState(t *testing.T) { + tests := []struct { + name string + event string + want domain.ActivityState + wantOK bool + }{ + {"user prompt -> active", "user-prompt-submit", domain.ActivityActive, 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}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := DeriveActivityState(tt.event, []byte(`{}`)) + if got != tt.want || ok != tt.wantOK { + t.Fatalf("DeriveActivityState(%q) = (%q, %v), want (%q, %v)", + tt.event, got, ok, tt.want, tt.wantOK) + } + }) + } +} diff --git a/backend/internal/adapters/agent/codex/codex.go b/backend/internal/adapters/agent/codex/codex.go index 874bef03f..a9a32a11c 100644 --- a/backend/internal/adapters/agent/codex/codex.go +++ b/backend/internal/adapters/agent/codex/codex.go @@ -61,8 +61,9 @@ func (p *Plugin) GetConfigSpec(ctx context.Context) (ports.ConfigSpec, error) { } // GetLaunchCommand builds the argv to start a new Codex session, applying the -// no-update-check and approval flags, optional system-prompt instructions, and -// the initial prompt (passed after `--` so a leading "-" is not read as a flag). +// no-update-check, hook-trust bypass, and approval flags, optional +// system-prompt instructions, and the initial prompt (passed after `--` so a +// leading "-" is not read as a flag). func (p *Plugin) GetLaunchCommand(ctx context.Context, cfg ports.LaunchConfig) (cmd []string, err error) { binary, err := p.codexBinary(ctx) if err != nil { @@ -71,6 +72,7 @@ func (p *Plugin) GetLaunchCommand(ctx context.Context, cfg ports.LaunchConfig) ( cmd = []string{binary} appendNoUpdateCheckFlag(&cmd) + appendHookTrustBypassFlag(&cmd) appendApprovalFlags(&cmd, cfg.Permissions) if cfg.SystemPromptFile != "" { @@ -116,6 +118,7 @@ func (p *Plugin) GetRestoreCommand(ctx context.Context, cfg ports.RestoreConfig) cmd = make([]string, 0, 8) cmd = append(cmd, binary, "resume") appendNoUpdateCheckFlag(&cmd) + appendHookTrustBypassFlag(&cmd) appendApprovalFlags(&cmd, cfg.Permissions) cmd = append(cmd, agentSessionID) return cmd, true, nil @@ -227,6 +230,14 @@ func appendNoUpdateCheckFlag(cmd *[]string) { *cmd = append(*cmd, "-c", "check_for_update_on_startup=false") } +func appendHookTrustBypassFlag(cmd *[]string) { + // AO installs deterministic workspace-local Codex hooks immediately before + // launch/restore. Without this flag, a fresh per-session worktree can skip + // those hooks until an interactive /hooks trust review happens, leaving AO + // without activity signals. + *cmd = append(*cmd, "--dangerously-bypass-hook-trust") +} + func appendApprovalFlags(cmd *[]string, permissions ports.PermissionMode) { switch normalizePermissionMode(permissions) { case ports.PermissionModeDefault: diff --git a/backend/internal/adapters/agent/codex/codex_test.go b/backend/internal/adapters/agent/codex/codex_test.go index 6547720a3..3d93c9dd7 100644 --- a/backend/internal/adapters/agent/codex/codex_test.go +++ b/backend/internal/adapters/agent/codex/codex_test.go @@ -28,6 +28,7 @@ func TestGetLaunchCommandBuildsCrossPlatformArgv(t *testing.T) { want := []string{ "codex", "-c", "check_for_update_on_startup=false", + "--dangerously-bypass-hook-trust", "--dangerously-bypass-approvals-and-sandbox", "-c", "model_instructions_file=" + filepath.Join("tmp", "prompt with spaces.md"), "--", "-fix this", @@ -249,6 +250,7 @@ func TestGetRestoreCommandReadsAgentSessionID(t *testing.T) { "codex", "resume", "-c", "check_for_update_on_startup=false", + "--dangerously-bypass-hook-trust", "--ask-for-approval", "on-request", "-c", `approvals_reviewer="auto_review"`, "thread-123", diff --git a/backend/internal/adapters/agent/codex/hooks.go b/backend/internal/adapters/agent/codex/hooks.go index 9d19609ac..5470b7bec 100644 --- a/backend/internal/adapters/agent/codex/hooks.go +++ b/backend/internal/adapters/agent/codex/hooks.go @@ -34,7 +34,7 @@ type codexHookFile struct { } type codexMatcherGroup struct { - Matcher *string `json:"matcher"` + Matcher *string `json:"matcher,omitempty"` Hooks []codexHookEntry `json:"hooks"` } @@ -56,6 +56,7 @@ type codexHookSpec struct { var codexManagedHooks = []codexHookSpec{ {Event: "SessionStart", Command: codexHookCommandPrefix + "session-start"}, {Event: "UserPromptSubmit", Command: codexHookCommandPrefix + "user-prompt-submit"}, + {Event: "PermissionRequest", Command: codexHookCommandPrefix + "permission-request"}, {Event: "Stop", Command: codexHookCommandPrefix + "stop"}, } diff --git a/backend/internal/adapters/agent/opencode/activity.go b/backend/internal/adapters/agent/opencode/activity.go new file mode 100644 index 000000000..e9ad7d3b4 --- /dev/null +++ b/backend/internal/adapters/agent/opencode/activity.go @@ -0,0 +1,18 @@ +package opencode + +import "github.com/aoagents/agent-orchestrator/backend/internal/domain" + +// DeriveActivityState maps an opencode plugin hook event onto an AO activity +// state. The bool is false when the event carries no activity signal. +func DeriveActivityState(event string, _ []byte) (domain.ActivityState, bool) { + switch event { + case "session-start": + return domain.ActivityActive, true + case "user-prompt-submit": + return domain.ActivityActive, true + case "stop": + return domain.ActivityIdle, true + default: + return "", false + } +} diff --git a/backend/internal/adapters/agent/opencode/activity_test.go b/backend/internal/adapters/agent/opencode/activity_test.go new file mode 100644 index 000000000..e40d23617 --- /dev/null +++ b/backend/internal/adapters/agent/opencode/activity_test.go @@ -0,0 +1,30 @@ +package opencode + +import ( + "testing" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" +) + +func TestDeriveActivityState(t *testing.T) { + tests := []struct { + name string + event string + want domain.ActivityState + wantOK bool + }{ + {"session start -> active", "session-start", domain.ActivityActive, true}, + {"user prompt -> active", "user-prompt-submit", domain.ActivityActive, true}, + {"stop -> idle", "stop", domain.ActivityIdle, true}, + {"unknown event -> no signal", "frobnicate", "", false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := DeriveActivityState(tt.event, []byte(`{}`)) + if got != tt.want || ok != tt.wantOK { + t.Fatalf("DeriveActivityState(%q) = (%q, %v), want (%q, %v)", + tt.event, got, ok, tt.want, tt.wantOK) + } + }) + } +} diff --git a/backend/internal/cli/client.go b/backend/internal/cli/client.go index c4e405f5c..92eafd441 100644 --- a/backend/internal/cli/client.go +++ b/backend/internal/cli/client.go @@ -93,7 +93,7 @@ func (c *commandContext) doJSON(ctx context.Context, method, path string, body, reader = bytes.NewReader(payload) } url := fmt.Sprintf("http://%s:%d/api/v1/%s", config.LoopbackHost, info.Port, path) - req, err := http.NewRequestWithContext(ctx, method, url, reader) + req, err := http.NewRequestWithContext(ctx, method, url, reader) // #nosec G704 -- daemon host is fixed loopback; path is an internal API route. if err != nil { return err } @@ -105,7 +105,7 @@ func (c *commandContext) doJSON(ctx context.Context, method, path string, body, // give daemon API calls far more headroom than the 2s status-probe timeout. client := *c.deps.HTTPClient client.Timeout = commandTimeout - resp, err := client.Do(req) + resp, err := client.Do(req) // #nosec G704 -- request target is the fixed loopback daemon URL above. if err != nil { return fmt.Errorf("call daemon: %w", err) } diff --git a/backend/internal/cli/hooks.go b/backend/internal/cli/hooks.go new file mode 100644 index 000000000..302b274d3 --- /dev/null +++ b/backend/internal/cli/hooks.go @@ -0,0 +1,71 @@ +package cli + +import ( + "context" + "fmt" + "io" + "net/url" + "os" + "strings" + + "github.com/spf13/cobra" + + "github.com/aoagents/agent-orchestrator/backend/internal/adapters/agent/activitydispatch" +) + +// 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. +type setActivityAPIRequest struct { + State string `json:"state"` +} + +// newHooksCommand builds the hidden `ao hooks ` command that +// agent CLIs invoke from their workspace-local hook config. It reads the native +// hook payload from stdin and the AO session id from AO_SESSION_ID, derives an +// activity state for the event, and reports it to the daemon. +// +// It is best-effort by design: a hook must never break the user's agent, so a +// non-AO session (no AO_SESSION_ID), an event that carries no activity signal, +// or an unreachable daemon all exit 0 rather than erroring. +func newHooksCommand(ctx *commandContext) *cobra.Command { + return &cobra.Command{ + Use: "hooks ", + Short: "Receive an agent hook callback (internal)", + Hidden: true, + Args: cobra.ExactArgs(2), + RunE: func(cmd *cobra.Command, args []string) error { + return ctx.runHook(cmd.Context(), args[0], args[1]) + }, + } +} + +func (c *commandContext) runHook(ctx context.Context, agent, event string) error { + sessionID := strings.TrimSpace(os.Getenv("AO_SESSION_ID")) + if sessionID == "" { + // Not an AO-managed session. Return before reading stdin so a manual + // invocation without a piped payload can't block on EOF. + return nil + } + payload, err := io.ReadAll(c.deps.In) + if err != nil { + // Surface read errors to stderr for parity with the daemon-error path, + // but keep the empty payload and exit 0: a failed hook must not break + // the agent. The deriver tolerates an empty payload. + _, _ = fmt.Fprintf(c.deps.Err, "ao hooks %s %s: read stdin: %v\n", agent, event, err) + } + + state, ok := activitydispatch.Derive(agent, event, payload) + if !ok { + // Unknown agent, or an event that carries no activity signal: report nothing. + return nil + } + + path := "sessions/" + url.PathEscape(sessionID) + "/activity" + if err := c.postJSON(ctx, path, setActivityAPIRequest{State: string(state)}, nil); err != nil { + // Report to stderr (the agent's hook runner captures it) for diagnosis, + // but exit 0: a failed activity report must not disrupt the agent. + _, _ = fmt.Fprintf(c.deps.Err, "ao hooks %s %s: %v\n", agent, event, err) + } + return nil +} diff --git a/backend/internal/cli/hooks_test.go b/backend/internal/cli/hooks_test.go new file mode 100644 index 000000000..45f571e3c --- /dev/null +++ b/backend/internal/cli/hooks_test.go @@ -0,0 +1,212 @@ +package cli + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +type activityCapture struct { + body string + path string + hits int +} + +// activityServer accepts POST /api/v1/sessions/{id}/activity and records what +// the CLI sent. It mirrors sendServer in send_test.go. +func activityServer(t *testing.T, status int, respBody string) (*httptest.Server, *activityCapture) { + t.Helper() + capture := &activityCapture{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || !strings.HasSuffix(r.URL.Path, "/activity") { + http.NotFound(w, r) + return + } + body, err := io.ReadAll(r.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + capture.body = string(body) + capture.path = r.URL.Path + capture.hits++ + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = io.WriteString(w, respBody) + })) + t.Cleanup(srv.Close) + return srv, capture +} + +func capturedState(t *testing.T, capture *activityCapture) string { + t.Helper() + var req struct { + State string `json:"state"` + } + if err := json.Unmarshal([]byte(capture.body), &req); err != nil { + t.Fatalf("decode body: %v\nbody=%s", err, capture.body) + } + return req.State +} + +func TestHooks_NotificationReportsWaitingInput(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"}`) + writeRunFileFor(t, cfg, srv) + + _, errOut, err := executeCLI(t, Deps{ + In: strings.NewReader(`{"notification_type":"idle_prompt"}`), + ProcessAlive: func(int) bool { return true }, + }, "hooks", "claude-code", "notification") + if err != nil { + t.Fatalf("unexpected error: %v\nstderr=%s", err, errOut) + } + 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) + } +} + +func TestHooks_SessionEndReportsExited(t *testing.T) { + 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(`{"reason":"logout"}`), + ProcessAlive: func(int) bool { return true }, + }, "hooks", "claude-code", "session-end") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := capturedState(t, capture); got != "exited" { + t.Errorf("state = %q, want exited", got) + } +} + +func TestHooks_StopReportsIdle(t *testing.T) { + 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(`{}`), + ProcessAlive: func(int) bool { return true }, + }, "hooks", "claude-code", "stop") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := capturedState(t, capture); got != "idle" { + t.Errorf("state = %q, want idle", got) + } +} + +func TestHooks_CodexPermissionRequestReportsWaitingInput(t *testing.T) { + 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", "codex", "permission-request") + 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) + } +} + +func TestHooks_OpenCodeUserPromptReportsActive(t *testing.T) { + 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(`{"session_id":"ses-1","prompt":"fix this"}`), + ProcessAlive: func(int) bool { return true }, + }, "hooks", "opencode", "user-prompt-submit") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := capturedState(t, capture); got != "active" { + t.Errorf("state = %q, want active", got) + } +} + +func TestHooks_NoSessionIDIsNoOp(t *testing.T) { + t.Setenv("AO_SESSION_ID", "") + cfg := setConfigEnv(t) + srv, capture := activityServer(t, http.StatusOK, `{}`) + writeRunFileFor(t, cfg, srv) + + _, _, err := executeCLI(t, Deps{ + In: strings.NewReader(`{"notification_type":"idle_prompt"}`), + ProcessAlive: func(int) bool { return true }, + }, "hooks", "claude-code", "notification") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if capture.hits != 0 { + t.Errorf("expected no daemon call for a non-AO session, got %d", capture.hits) + } +} + +func TestHooks_UntrackedEventIsNoOp(t *testing.T) { + t.Setenv("AO_SESSION_ID", "ao-7") + cfg := setConfigEnv(t) + srv, capture := activityServer(t, http.StatusOK, `{}`) + writeRunFileFor(t, cfg, srv) + + _, _, err := executeCLI(t, Deps{ + In: strings.NewReader(`{"notification_type":"auth_success"}`), + ProcessAlive: func(int) bool { return true }, + }, "hooks", "claude-code", "notification") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if capture.hits != 0 { + t.Errorf("expected no daemon call for an untracked notification, got %d", capture.hits) + } +} + +func TestHooks_DaemonDownIsBestEffort(t *testing.T) { + t.Setenv("AO_SESSION_ID", "ao-7") + setConfigEnv(t) // no run-file written: daemon is "not running" + + _, _, err := executeCLI(t, Deps{ + In: strings.NewReader(`{"reason":"logout"}`), + }, "hooks", "claude-code", "session-end") + if err != nil { + t.Fatalf("hooks must be best-effort (exit 0) when the daemon is down, got: %v", err) + } +} + +func TestHooks_DaemonErrorIsSwallowed(t *testing.T) { + t.Setenv("AO_SESSION_ID", "ao-7") + cfg := setConfigEnv(t) + srv, _ := activityServer(t, http.StatusInternalServerError, + `{"error":"internal","code":"BOOM","message":"boom"}`) + writeRunFileFor(t, cfg, srv) + + _, errOut, err := executeCLI(t, Deps{ + In: strings.NewReader(`{"reason":"logout"}`), + ProcessAlive: func(int) bool { return true }, + }, "hooks", "claude-code", "session-end") + if err != nil { + t.Fatalf("hooks must exit 0 even on a daemon error, got: %v", err) + } + if !strings.Contains(errOut, "ao hooks") { + t.Errorf("expected the failure surfaced to stderr, got %q", errOut) + } +} diff --git a/backend/internal/cli/root.go b/backend/internal/cli/root.go index daabd5b81..4dec629c2 100644 --- a/backend/internal/cli/root.go +++ b/backend/internal/cli/root.go @@ -167,6 +167,7 @@ func NewRootCommand(deps Deps) *cobra.Command { root.AddCommand(newDoctorCommand(ctx)) root.AddCommand(newSpawnCommand(ctx)) root.AddCommand(newSendCommand(ctx)) + root.AddCommand(newHooksCommand(ctx)) root.AddCommand(newProjectCommand(ctx)) root.AddCommand(newSessionCommand(ctx)) root.AddCommand(newOrchestratorCommand(ctx)) diff --git a/backend/internal/daemon/daemon.go b/backend/internal/daemon/daemon.go index 58557abf9..983b95ace 100644 --- a/backend/internal/daemon/daemon.go +++ b/backend/internal/daemon/daemon.go @@ -107,6 +107,7 @@ func Run() error { Sessions: sessionSvc, CDC: store, Events: cdcPipe.Broadcaster, + Activity: lcStack.LCM, }) if err != nil { stop() diff --git a/backend/internal/httpd/api.go b/backend/internal/httpd/api.go index 63dd277cf..f4e683853 100644 --- a/backend/internal/httpd/api.go +++ b/backend/internal/httpd/api.go @@ -19,6 +19,7 @@ import ( type APIDeps struct { Projects projectsvc.Manager Sessions controllers.SessionService + Activity controllers.ActivityRecorder PRs prsvc.ActionManager CDC cdc.Source Events cdcSubscriber @@ -44,7 +45,8 @@ func NewAPI(cfg config.Config, deps APIDeps) *API { Mgr: deps.Projects, }, sessions: &controllers.SessionsController{ - Svc: deps.Sessions, + Svc: deps.Sessions, + Activity: deps.Activity, }, prs: &controllers.PRsController{Svc: deps.PRs}, events: &EventsController{Source: deps.CDC, Live: deps.Events}, diff --git a/backend/internal/httpd/apispec/openapi.yaml b/backend/internal/httpd/apispec/openapi.yaml index d925a2214..7a5c6f5c8 100644 --- a/backend/internal/httpd/apispec/openapi.yaml +++ b/backend/internal/httpd/apispec/openapi.yaml @@ -544,6 +544,57 @@ paths: summary: Rename a session display name tags: - sessions + /api/v1/sessions/{sessionId}/activity: + post: + operationId: setSessionActivity + parameters: + - description: Session identifier, e.g. project-1. + in: path + name: sessionId + required: true + schema: + description: Session identifier, e.g. project-1. + type: string + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/SetActivityRequest' + required: true + responses: + "200": + content: + application/json: + schema: + $ref: '#/components/schemas/SetActivityResponse' + description: OK + "400": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Bad Request + "404": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Not Found + "500": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Internal Server Error + "501": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Not Implemented + summary: Report an agent activity-state signal for a session + tags: + - sessions /api/v1/sessions/{sessionId}/kill: post: operationId: killSession @@ -1227,6 +1278,32 @@ components: required: - session type: object + SetActivityRequest: + properties: + state: + description: Agent activity state reported by an agent hook. + enum: + - active + - idle + - waiting_input + - exited + type: string + required: + - state + type: object + SetActivityResponse: + properties: + ok: + type: boolean + sessionId: + type: string + state: + type: string + required: + - ok + - sessionId + - state + type: object SpawnOrchestratorRequest: properties: clean: diff --git a/backend/internal/httpd/apispec/specgen/build.go b/backend/internal/httpd/apispec/specgen/build.go index 824e227e3..c7dbc7586 100644 --- a/backend/internal/httpd/apispec/specgen/build.go +++ b/backend/internal/httpd/apispec/specgen/build.go @@ -143,6 +143,8 @@ var schemaNames = map[string]string{ "ControllersClaimPRRequest": "ClaimPRRequest", "ControllersSessionPRFacts": "SessionPRFacts", "ControllersListSessionPRsResponse": "ListSessionPRsResponse", + "ControllersSetActivityRequest": "SetActivityRequest", + "ControllersSetActivityResponse": "SetActivityResponse", "ControllersSpawnOrchestratorRequest": "SpawnOrchestratorRequest", "ControllersSpawnOrchestratorResponse": "SpawnOrchestratorResponse", "ControllersOrchestratorResponse": "OrchestratorResponse", @@ -424,6 +426,19 @@ func sessionOperations() []operation { {http.StatusInternalServerError, envelope.APIError{}}, }, }, + { + method: http.MethodPost, path: "/api/v1/sessions/{sessionId}/activity", id: "setSessionActivity", tag: "sessions", + summary: "Report an agent activity-state signal for a session", + pathParams: []any{controllers.SessionIDParam{}}, + reqBody: controllers.SetActivityRequest{}, + resps: []respUnit{ + {http.StatusOK, controllers.SetActivityResponse{}}, + {http.StatusBadRequest, envelope.APIError{}}, + {http.StatusNotFound, envelope.APIError{}}, + {http.StatusInternalServerError, envelope.APIError{}}, + {http.StatusNotImplemented, envelope.APIError{}}, + }, + }, { method: http.MethodGet, path: "/api/v1/orchestrators", id: "listOrchestrators", tag: "sessions", summary: "List orchestrator sessions across projects", diff --git a/backend/internal/httpd/controllers/dto.go b/backend/internal/httpd/controllers/dto.go index d35aaa943..f0fcbe975 100644 --- a/backend/internal/httpd/controllers/dto.go +++ b/backend/internal/httpd/controllers/dto.go @@ -209,6 +209,18 @@ type ClaimPRResponse struct { TakenOverFrom []domain.SessionID `json:"takenOverFrom"` } +// SetActivityRequest is the body of POST /api/v1/sessions/{sessionId}/activity. +type SetActivityRequest struct { + State string `json:"state" enum:"active,idle,waiting_input,exited" description:"Agent activity state reported by an agent hook."` +} + +// SetActivityResponse is the body of POST /api/v1/sessions/{sessionId}/activity. +type SetActivityResponse struct { + OK bool `json:"ok"` + SessionID domain.SessionID `json:"sessionId"` + State string `json:"state"` +} + // OrchestratorIDParam is the {id} path parameter for orchestrator routes. type OrchestratorIDParam struct { ID string `path:"id" description:"Orchestrator session identifier, e.g. project-orchestrator."` diff --git a/backend/internal/httpd/controllers/sessions.go b/backend/internal/httpd/controllers/sessions.go index a42e8f98c..50a045447 100644 --- a/backend/internal/httpd/controllers/sessions.go +++ b/backend/internal/httpd/controllers/sessions.go @@ -37,10 +37,16 @@ type SessionService interface { ClaimPR(ctx context.Context, id domain.SessionID, ref string, opts sessionsvc.ClaimPROptions) (sessionsvc.ClaimPRResult, error) } +// ActivityRecorder applies an agent activity-state signal to a session. +type ActivityRecorder interface { + ApplyActivitySignal(ctx context.Context, id domain.SessionID, s ports.ActivitySignal) error +} + // SessionsController owns the session routes. Nil keeps routes registered but // returns OpenAPI-backed 501s. type SessionsController struct { - Svc SessionService + Svc SessionService + Activity ActivityRecorder } // Register mounts the session routes on the supplied router. @@ -55,6 +61,7 @@ func (c *SessionsController) Register(r chi.Router) { r.Post("/sessions/{sessionId}/restore", c.restore) r.Post("/sessions/{sessionId}/kill", c.kill) r.Post("/sessions/{sessionId}/send", c.send) + r.Post("/sessions/{sessionId}/activity", c.activity) r.Get("/orchestrators", c.listOrchestrators) r.Post("/orchestrators", c.spawnOrchestrator) r.Get("/orchestrators/{id}", c.getOrchestrator) @@ -246,6 +253,30 @@ func (c *SessionsController) send(w http.ResponseWriter, r *http.Request) { envelope.WriteJSON(w, http.StatusOK, SendSessionMessageResponse{OK: true, SessionID: sessionID(r), Message: message}) } +func (c *SessionsController) activity(w http.ResponseWriter, r *http.Request) { + if c.Activity == nil { + apispec.NotImplemented(w, r, "POST", "/api/v1/sessions/{sessionId}/activity") + return + } + var in SetActivityRequest + if err := decodeJSON(r, &in); err != nil { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "INVALID_JSON", "Invalid JSON body", nil) + return + } + state := domain.ActivityState(in.State) + switch state { + case domain.ActivityActive, domain.ActivityIdle, domain.ActivityWaitingInput, 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 { + envelope.WriteError(w, r, err) + return + } + envelope.WriteJSON(w, http.StatusOK, SetActivityResponse{OK: true, SessionID: sessionID(r), State: in.State}) +} + 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 new file mode 100644 index 000000000..ed63dc4b5 --- /dev/null +++ b/backend/internal/httpd/controllers/sessions_activity_test.go @@ -0,0 +1,99 @@ +package controllers_test + +import ( + "context" + "errors" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "github.com/aoagents/agent-orchestrator/backend/internal/config" + "github.com/aoagents/agent-orchestrator/backend/internal/domain" + "github.com/aoagents/agent-orchestrator/backend/internal/httpd" + "github.com/aoagents/agent-orchestrator/backend/internal/ports" +) + +type fakeActivityRecorder struct { + gotID domain.SessionID + gotSignal ports.ActivitySignal + calls int + err error +} + +func (f *fakeActivityRecorder) ApplyActivitySignal(_ context.Context, id domain.SessionID, s ports.ActivitySignal) error { + f.calls++ + f.gotID = id + f.gotSignal = s + return f.err +} + +func newActivityTestServer(t *testing.T, rec *fakeActivityRecorder) *httptest.Server { + t.Helper() + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + deps := httpd.APIDeps{} + if rec != nil { + deps.Activity = rec + } + srv := httptest.NewServer(httpd.NewRouterWithControl(config.Config{}, log, nil, deps, httpd.ControlDeps{})) + t.Cleanup(srv.Close) + return srv +} + +func TestSessionsAPI_ActivityAppliesSignal(t *testing.T) { + rec := &fakeActivityRecorder{} + srv := newActivityTestServer(t, rec) + + body, status, _ := doRequest(t, srv, "POST", "/api/v1/sessions/ao-1/activity", `{"state":"waiting_input"}`) + if status != http.StatusOK { + t.Fatalf("activity = %d, want 200; body=%s", status, body) + } + var resp struct { + OK bool `json:"ok"` + SessionID string `json:"sessionId"` + State string `json:"state"` + } + mustJSON(t, body, &resp) + if !resp.OK || resp.SessionID != "ao-1" || resp.State != "waiting_input" { + t.Fatalf("activity response = %#v", resp) + } + if rec.calls != 1 || rec.gotID != "ao-1" { + t.Fatalf("recorder calls=%d id=%q", rec.calls, rec.gotID) + } + if !rec.gotSignal.Valid || rec.gotSignal.State != domain.ActivityWaitingInput { + t.Fatalf("recorder signal = %#v", rec.gotSignal) + } +} + +func TestSessionsAPI_ActivityRejectsUnknownState(t *testing.T) { + rec := &fakeActivityRecorder{} + srv := newActivityTestServer(t, rec) + + body, status, _ := doRequest(t, srv, "POST", "/api/v1/sessions/ao-1/activity", `{"state":"napping"}`) + assertErrorCode(t, body, status, http.StatusBadRequest, "INVALID_ACTIVITY_STATE") + if rec.calls != 0 { + t.Fatalf("recorder should not be called for an invalid state; calls=%d", rec.calls) + } +} + +func TestSessionsAPI_ActivityRejectsBadJSON(t *testing.T) { + srv := newActivityTestServer(t, &fakeActivityRecorder{}) + + body, status, _ := doRequest(t, srv, "POST", "/api/v1/sessions/ao-1/activity", `{`) + assertErrorCode(t, body, status, http.StatusBadRequest, "INVALID_JSON") +} + +func TestSessionsAPI_ActivityRecorderErrorIs500(t *testing.T) { + srv := newActivityTestServer(t, &fakeActivityRecorder{err: errors.New("boom")}) + + body, status, _ := doRequest(t, srv, "POST", "/api/v1/sessions/ao-1/activity", `{"state":"exited"}`) + assertErrorCode(t, body, status, http.StatusInternalServerError, "INTERNAL_ERROR") +} + +func TestSessionsAPI_ActivityWithoutRecorderIs501(t *testing.T) { + srv := newActivityTestServer(t, nil) + + body, status, _ := doRequest(t, srv, "POST", "/api/v1/sessions/ao-1/activity", `{"state":"idle"}`) + assertErrorCode(t, body, status, http.StatusNotImplemented, "NOT_IMPLEMENTED") +} diff --git a/backend/internal/lifecycle/manager.go b/backend/internal/lifecycle/manager.go index 8b7eb3f26..d0bd9c24d 100644 --- a/backend/internal/lifecycle/manager.go +++ b/backend/internal/lifecycle/manager.go @@ -127,8 +127,13 @@ func (m *Manager) MarkTerminated(ctx context.Context, id domain.SessionID) error }) } +// sameActivity reports whether two activity signals describe the same state. +// LastActivityAt is intentionally ignored: same-state repeats (e.g. a stream +// of idle notifications) must not rewrite UpdatedAt or fan out a CDC event. +// LastActivityAt now marks when this state was first entered since the last +// transition, which is the timestamp a UI actually wants. func sameActivity(a, b domain.Activity) bool { - return a.State == b.State && a.LastActivityAt.Equal(b.LastActivityAt) + return a.State == b.State } func mergeMetadata(base, in domain.SessionMetadata) domain.SessionMetadata { diff --git a/frontend/src/api/schema.ts b/frontend/src/api/schema.ts index f915e5e21..764a47878 100644 --- a/frontend/src/api/schema.ts +++ b/frontend/src/api/schema.ts @@ -162,6 +162,23 @@ export interface paths { patch: operations["renameSession"]; trace?: never; }; + "/api/v1/sessions/{sessionId}/activity": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** Report an agent activity-state signal for a session */ + post: operations["setSessionActivity"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/v1/sessions/{sessionId}/kill": { parameters: { query?: never; @@ -432,6 +449,18 @@ export interface components { SessionResponse: { session: components["schemas"]["Session"]; }; + SetActivityRequest: { + /** + * @description Agent activity state reported by an agent hook. + * @enum {string} + */ + state: "active" | "idle" | "waiting_input" | "exited"; + }; + SetActivityResponse: { + ok: boolean; + sessionId: string; + state: string; + }; SpawnOrchestratorRequest: { clean?: boolean; projectId: string; @@ -1144,6 +1173,69 @@ export interface operations { }; }; }; + setSessionActivity: { + parameters: { + query?: never; + header?: never; + path: { + /** @description Session identifier, e.g. project-1. */ + sessionId: string; + }; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["SetActivityRequest"]; + }; + }; + responses: { + /** @description OK */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["SetActivityResponse"]; + }; + }; + /** @description Bad Request */ + 400: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Not Found */ + 404: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Internal Server Error */ + 500: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Not Implemented */ + 501: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + }; + }; killSession: { parameters: { query?: never;