diff --git a/docs/logs/INDEX.md b/docs/logs/INDEX.md index 78d08e4d..23345437 100644 --- a/docs/logs/INDEX.md +++ b/docs/logs/INDEX.md @@ -90,3 +90,6 @@ PR #1113 is merged to `main` as the #1106 rebase baseline. Issue #1106 documents mixed-version callback dispatch fencing, bootstrap-only crash recovery provenance, and eventual pre-claim contention progress. + +Recent entry: Issue #1009 additive server task lifecycle projection and native +macOS controls/reconciliation (pending final repository regression at time of entry). diff --git a/docs/logs/engineering-log.md b/docs/logs/engineering-log.md index def88a91..f27baac0 100644 --- a/docs/logs/engineering-log.md +++ b/docs/logs/engineering-log.md @@ -4329,3 +4329,71 @@ Skipped creating separate issues for Op/EventMsg protocol (already covered by SS `./scripts/test-regression.sh` rerun passed normal, race, and coverage at 85.5% with zero uncovered functions. This is not PTY or native-GUI proof; those remain the #1000 convergence matrix. + +## 2026-08-04 (Issue #1009 — scheduled-task lifecycle and macOS controls) + +- Change: `GET /v1/tasks` now projects optional, server-authored cron and + callback lifecycle fields: conversation linkage, cron next/last timestamps, + most-recent execution state/run/error, callback due time, and update time. + Existing type-specific routes remain the sole mutation authority. +- Native app: `TaskInfo` now has typed forward-compatible kind, state, and + action values; unknown server values decode without making the Activity page + unusable. HarnessKit adds scoped pause/resume/delete/cancel requests. + Activity displays lifecycle detail and accessible controls, asks before cron + deletion, and always reloads server state after an action succeeds or fails. +- TDD: the first Go test failed because `Task` had no lifecycle fields; the + first Swift test failed because task values were raw strings and control APIs + were absent. A full Swift run then caught a global URLProtocol-stub race in + the new tests; the task tests now use their own isolated protocol class. +- Verification: focused task lifecycle tests, `go test ./internal/server + -count=1`, `go test ./internal/server -race -count=1`, and the full Swift + suite (`256` tests) passed. The direct full repository gate, run with its own + temporary cache and coverage profile after rebase to `f7b6c70`, passed normal, + race, and coverage phases at `85.5%` total coverage with zero uncovered + functions. This makes the implementation ready for review; it is not the + separate #1010 API/TUI/native full-conversation proof. +- Review repair: cron Activity actions now carry optional `expected_updated_at` + only when the row provides it; server actions preserve empty legacy bodies, + require active-to-pause and paused-to-resume state, and map stale/invalid + mutations to 409 without changing the job. Callback `updated_at` is read + from the durable row on every list/get/returning path and projects into the + task row. The native "Open linked run" control opens the durable conversation + and lets only non-terminal run-event reducer evidence establish live control + ownership; terminal and missing links cannot manufacture controls. Repair + regressions cover stale/current actions, persisted callback freshness, JSON + request shape, and active/terminal/missing navigation. +- Repair verification: complete affected server/tool normal and race suites + passed; the full native suite passed 259 tests. The first full repository + attempt ran concurrently with another coverage regression and hit two + unrelated `cmd/harnessd` three-second startup timeouts, so it was not + accepted. After that load completed, the serial rerun with fresh cache/profile + passed normal, race, and coverage at 85.5% total with zero uncovered + functions. + +## 2026-08-04 (Issue #1009 review repair — opaque cron task version) + +- Cause: HarnessKit decoded task `updated_at` into `Date`, then encoded it + with a new ISO-8601 formatter for `expected_updated_at`. That conversion can + discard server-issued nanoseconds, making an otherwise fresh Activity row + fail its cron CAS action with 409. +- Fix: `TaskInfo.updatedAtVersion` retains the raw optional `updated_at` + string through ProjectSession and `TaskActionVersion`; standard `Encodable` + now emits the token unchanged. Missing versions still use the existing empty + request body for older additive task payloads. +- Regression: Swift asserts a `.123456789Z` task token and the exact JSON + action field; Go lists a nanosecond cron token, proves `.123Z` returns 409 + without mutation, then proves the exact listed token pauses it. + +## 2026-08-04 (Issue #1009 review repair — no-store callback terminals) + +- Cause: no-store callback terminal rows stay in `m.callbacks` but leave the + active `byConv` index. `ListAllCallbacks` incorrectly walked that active + index, making canceled/fired/shutdown callbacks vanish from `/v1/tasks`. +- Fix: all-state listing snapshots and safely projects `m.callbacks`; legacy + conversation `List`/`ListCallbacks` remain active-only through `byConv`. + Legacy cancel, fire, and shutdown cancellation now stamp `UpdatedAt` from + the manager clock. The durable cancel/list branches are unchanged. +- Regression: deterministic manager coverage proves terminal timestamps and + all-state retention while agent-facing lists exclude terminals; server + coverage proves cancel then `GET /v1/tasks` returns one canceled read-only + row with nonzero `updated_at`. diff --git a/docs/plans/2026-08-04-issue-1009-macapp-task-lifecycle-impact-map.md b/docs/plans/2026-08-04-issue-1009-macapp-task-lifecycle-impact-map.md new file mode 100644 index 00000000..1e196984 --- /dev/null +++ b/docs/plans/2026-08-04-issue-1009-macapp-task-lifecycle-impact-map.md @@ -0,0 +1,85 @@ +# Issue #1009 — Cross-Surface Impact Map + +## Task + +- Task / issue: #1009 macOS scheduled-task lifecycle and controls. +- Plan link: `2026-08-04-issue-1009-macapp-task-lifecycle-plan.md`. +- Owner: Codex implementation worktree. +- Status: in implementation. + +## Current Ownership, Callers, and Data Flow + +- Entry points: `internal/server/http_tasks.go` creates `Task`; existing + `http_cron.go` and callback action handler own mutation; macOS + `HarnessKit/ClientTasks.swift` decodes and `ActivityView.swift` renders. +- Source of truth: cron client/status/executions and callback manager state; + the server calculates task action availability. `ProjectSession.refreshActivity` + owns macOS reconciliation. +- Search evidence: `rg -n "TaskInfo|/v1/tasks|TaskAction|Activity" internal cmd macapp`. +- Conclusion: extend the common task DTO rather than duplicate a parallel + scheduled-task listing; task-specific fields remain optional. + +## Config, API, CLI, and Tools + +- No configuration, CLI command, tool schema, or provider routing change. +- Additive `/v1/tasks` fields: schedule timing, latest execution state/run/error + and callback due time; existing cron/callback endpoints remain action targets. +- Mac client sends optional row `updated_at` as its opaque raw server string in + scoped cron action JSON while preserving empty legacy action bodies; server maps stale/invalid action + state to 409. Errors are retained as `HarnessError` and no client-side + permission inference is trusted. + +## Persistence and Compatibility + +- No migration. Cron values originate from persisted cron storage; callback + `updated_at` is now selected/scanned from its existing durable column for + each lifecycle transition. +- Older server payloads omit all new optionals; unknown kind/state/action raw + values preserve display rather than causing decode failures. +- Mixed-version clients use existing coarse row fields and show no absent action. +- No-store callback managers retain terminal map rows for the all-state task + API, but their `byConv` compatibility index remains active-only so capacity + accounting and agent-facing conversation lists do not regress. + +## Lifecycle, Security, and Reliability + +- No new goroutines/timers. Activity polling remains bounded to the visible view. +- Existing runs:read/runs:write and tenant checks enforce authority. UI refreshes + after either action success or failure so stale action sets are not retained. + Cron pause/resume require the matching current state and an optional CAS + version; linked run controls require reducer-admitted live event evidence. +- Server-safe errors are surfaced; delete needs confirmation; cancel/pause/resume + depend only on advertised actions. + +## Product and Integration Surfaces + +- Server/runtime: task union gets lifecycle projection only. +- macOS: typed models, action client, task detail/controls, VoiceOver labels, + linked-run navigation/selection where available. +- TUI/web and external systems: none; existing API remains backward compatible. +- Provider/model/tool catalog: none; rendering never fabricates assistant output. + +## Deployment and Operations + +- Deploy server before clients (additive reads); rollback clients preserves + read-only task display. Roll back server fields without breaking clients. +- API task rows provide operator diagnostics; no new metrics or secrets. + +## Regression Tests + +- Red: task lifecycle projection and typed Swift action request/decode tests. +- New tests: cron active/paused lifecycle, exact nine-digit `updated_at` + list-to-action preservation plus stale truncated-token CAS rejection, + stale/current CAS action state, + no-store callback cancel/fire/shutdown terminal timestamp and all-state task + visibility with legacy active-list exclusion, + callback durable update-time projection/transition, versioned/legacy action + bodies, active/terminal/missing linked-run navigation, reconciliation after + success/failure, unknown task values, accessible labels. +- Exact commands: `go test ./internal/server -run 'TestTasks|TestCron' -count=1`, + race equivalent, `swift test --package-path macapp`, and full regression. + +## Documentation and Handoff + +- Update plan/log indexes and engineering log once implemented. No public + assertion of full native proof until #1010 uses an exact current artifact. diff --git a/docs/plans/2026-08-04-issue-1009-macapp-task-lifecycle-plan.md b/docs/plans/2026-08-04-issue-1009-macapp-task-lifecycle-plan.md new file mode 100644 index 00000000..cf0a2810 --- /dev/null +++ b/docs/plans/2026-08-04-issue-1009-macapp-task-lifecycle-plan.md @@ -0,0 +1,79 @@ +# Issue #1009 — macOS scheduled-task lifecycle and controls + +## Context + +- Governing GitHub issue: #1009 (child of #1000). +- Problem: `/v1/tasks` currently exposes only a generic task row and the macOS + Activity page renders it read-only. Cron jobs and callbacks therefore lack + visible timing/result/run linkage and cannot be controlled from the app. +- User impact: a deployment watcher must be observable, controllable, and + reconciled to the server rather than represented by a model text claim. +- Constraints: additive wire contract; server remains authoritative for every + action; do not alter scheduler or retry behavior; preserve generic task rows. + +## Scope + +- In scope: additive cron/callback lifecycle fields in task rows; typed, + forward-compatible Swift task values; client requests for pause/resume/delete + and cancel; Activity detail/action UX with accessibility; reconciliation and + regression coverage. +- Out of scope: scheduler persistence/retry algorithm, a general notification + system, TUI changes, and final #1010 live proof. + +## Documentation Contract + +- Feature status: in implementation. +- Public docs affected: native macOS and API behavior documentation only after + implementation is tested. +- Implementation notes: engineering log and plan indexes record the actual + additive contract. + +## Test Plan (TDD) + +- First red: server task-union contract asserts cron next/last execution/run + fields, callback due time, and authoritative actions; Swift client tests + assert typed unknown decoding and exact control requests. +- Existing tests updated: task API and Activity session tests. +- Review repair red: stale cron pause/resume/delete requests must return 409 + without mutation; versioned Swift requests must preserve an empty body for + older rows; active versus terminal linked runs must not share controls. +- Timestamp-wire repair red: a nanosecond `updated_at` token must decode and + return to the action endpoint byte-for-byte; a client-truncated token must + receive 409 without mutating the cron job. +- Regression: focused Go server normal/race, focused Swift package tests, full + `./scripts/test-regression.sh`, and `swift test --package-path macapp`. + +## Cross-Surface Impact Map + +See `2026-08-04-issue-1009-macapp-task-lifecycle-impact-map.md`. + +## Implementation Checklist + +- [x] Verify issue contract and architecture/search evidence. +- [x] Create plan and impact map. +- [x] Capture red task serialization/client-action tests. +- [x] Implement additive server and typed client contract. +- [x] Add accessible Activity actions and server reconciliation. +- [x] Update logs/indexes; issue evidence follows final regression. +- [x] Pass focused and full regression gates. +- [x] Preserve opaque cron action version tokens through final review repair. +- [x] Keep no-store callback terminal rows and their lifecycle timestamps in + the all-state task inventory without changing legacy active-only lists. + +## Risks and Mitigations + +- Stale UI actions could claim success: each action refreshes `/v1/tasks` in a + `defer` path and surfaces the server error. Cron actions additionally send + optional `expected_updated_at` as the opaque server string (rather than a + reformatted `Date`); stale versions and invalid state transitions return 409 + without mutation. +- Mixed server versions omit fields/actions: optional fields decode as absent, + unknown enum values remain displayable, and controls remain hidden. +- Destructive deletion: confirmation is required in the Activity UI. +- Linked run authority: opening a task loads its durable conversation first; + only a non-terminal event accepted by the run reducer can make its linked + run a live control target. +- No-store callback terminal visibility: `ListAllCallbacks` snapshots retained + manager rows for the task API, while `List` and `ListCallbacks` remain backed + by the active `byConv` index; every legacy cancel/fire/shutdown terminal + transition advances `updated_at`. diff --git a/docs/plans/INDEX.md b/docs/plans/INDEX.md index 3fdc3778..24897b64 100644 --- a/docs/plans/INDEX.md +++ b/docs/plans/INDEX.md @@ -165,6 +165,9 @@ - `2026-08-03-issue-1115-workflow-subscriber-plan.md` — Issue #1115 deterministic full-buffer workflow subscriber terminal-close regression plan. - `2026-08-03-issue-1115-workflow-subscriber-impact-map.md` — Cross-surface impact map for Issue #1115's test-only subscription scheduling repair. + +- `2026-08-04-issue-1009-macapp-task-lifecycle-plan.md` — Issue #1009 plan for additive scheduled-task lifecycle visibility and controls in the macOS app. +- `2026-08-04-issue-1009-macapp-task-lifecycle-impact-map.md` — Cross-surface impact map for #1009 server task projection and macOS reconciliation. - `2026-08-01-issue-1083-approval-publication-race-plan.md` — Issue #1083 plan for making approval-required publication an immediate-resolution readiness guarantee. - `2026-08-01-issue-1083-approval-publication-race-impact-map.md` — Cross-surface impact map for Issue #1083 broker registration ordering. diff --git a/internal/harness/tools/delayed_callback.go b/internal/harness/tools/delayed_callback.go index 2085f935..60b2089d 100644 --- a/internal/harness/tools/delayed_callback.go +++ b/internal/harness/tools/delayed_callback.go @@ -156,6 +156,7 @@ type CallbackInfo struct { State CallbackState `json:"state"` FiresAt time.Time `json:"fires_at"` CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` // TenantID and AgentID capture the originating run's scope so the fired // follow-up run is started on the same tenant + agent. Both may be empty // for the default/unscoped case. Omitted from JSON when empty to preserve @@ -323,6 +324,7 @@ func (m *CallbackManager) Set(req SetRequest) (CallbackInfo, error) { State: CallbackStatePending, FiresAt: now.Add(delay), CreatedAt: now, + UpdatedAt: now, TenantID: req.TenantID, AgentID: req.AgentID, Model: strings.TrimSpace(req.Model), @@ -400,6 +402,7 @@ func (m *CallbackManager) Cancel(id string) (CallbackInfo, error) { info := cb.info info.State = CallbackStateCanceled + info.UpdatedAt = m.now().UTC() if m.store != nil { if err := m.store.Update(context.Background(), info); err != nil { m.mu.Unlock() @@ -494,13 +497,12 @@ func (m *CallbackManager) ListAllCallbacks(ctx context.Context) ([]CallbackInfo, m.mu.Lock() defer m.mu.Unlock() + // The all-state API inventory includes retained terminal callbacks. The + // per-conversation active index intentionally omits those entries to free + // scheduling capacity, so it must not be used here. result := make([]CallbackInfo, 0, len(m.callbacks)) - for _, ids := range m.byConv { - for _, id := range ids { - if cb, ok := m.callbacks[id]; ok { - result = append(result, cb.info) - } - } + for _, cb := range m.callbacks { + result = append(result, publicCallbackInfo(cb.info)) } return result, nil } @@ -521,6 +523,7 @@ func (m *CallbackManager) Shutdown() { // Preserve the historical in-memory manager contract. Durable // managers retain pending rows for restart recovery instead. cb.info.State = CallbackStateCanceled + cb.info.UpdatedAt = m.now().UTC() m.removeFromByConv(cb.info.ConversationID, cb.info.ID) canceled = append(canceled, cb.info) } @@ -574,6 +577,7 @@ func (m *CallbackManager) fire(id string) { func (m *CallbackManager) fireLegacyLocked(id string, cb *pendingCallback) { info := cb.info info.State = CallbackStateFired + info.UpdatedAt = m.now().UTC() cb.info = info convID := info.ConversationID prompt := info.Prompt diff --git a/internal/harness/tools/delayed_callback_store.go b/internal/harness/tools/delayed_callback_store.go index 431ed6e4..442bad3b 100644 --- a/internal/harness/tools/delayed_callback_store.go +++ b/internal/harness/tools/delayed_callback_store.go @@ -299,11 +299,14 @@ func parseStoredCallbackTimeText(value string) (time.Time, error) { } func (s *SQLiteCallbackStore) Create(c context.Context, i CallbackInfo) error { i.LastError = SafeCallbackErrorSummary(i.LastError) + if i.UpdatedAt.IsZero() { + i.UpdatedAt = i.CreatedAt + } fallbackProviders, err := json.Marshal(i.FallbackProviders) if err != nil { return fmt.Errorf("marshal callback fallback providers: %w", err) } - _, e := s.db.ExecContext(c, `INSERT INTO delayed_callbacks(id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,updated_at,attempt,run_id,next_attempt_at,last_error)VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, i.ID, i.TenantID, i.AgentID, i.ConversationID, i.Prompt, i.Model, i.ProviderName, i.AllowFallback, string(fallbackProviders), i.Delay, i.FiresAt.UTC(), i.State, i.CreatedAt.UTC(), i.CreatedAt.UTC(), i.Attempt, i.RunID, nullableCallbackTime(i.NextAttemptAt), i.LastError) + _, e := s.db.ExecContext(c, `INSERT INTO delayed_callbacks(id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,updated_at,attempt,run_id,next_attempt_at,last_error)VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, i.ID, i.TenantID, i.AgentID, i.ConversationID, i.Prompt, i.Model, i.ProviderName, i.AllowFallback, string(fallbackProviders), i.Delay, i.FiresAt.UTC(), i.State, i.CreatedAt.UTC(), i.UpdatedAt.UTC(), i.Attempt, i.RunID, nullableCallbackTime(i.NextAttemptAt), i.LastError) return e } @@ -316,10 +319,10 @@ func nullableCallbackTime(value time.Time) any { // ClaimDue atomically fences one due pending/retry row for one dispatcher. func (s *SQLiteCallbackStore) ClaimDue(c context.Context, id, token string, now, until time.Time) (CallbackInfo, bool, error) { - return s.claimReturning(c, id, token, `UPDATE delayed_callbacks SET state='dispatching_fenced',next_attempt_at=NULL,dispatch_token=?,dispatch_lease_until=?,attempt=attempt+1,updated_at=? WHERE id=? AND ((state='pending' AND fires_at<=?) OR (state='retry_wait' AND next_attempt_at<=?)) RETURNING id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,run_id,attempt,next_attempt_at,last_error,dispatch_token,dispatch_lease_until`, token, until.UTC(), now.UTC(), id, now.UTC(), now.UTC()) + return s.claimReturning(c, id, token, `UPDATE delayed_callbacks SET state='dispatching_fenced',next_attempt_at=NULL,dispatch_token=?,dispatch_lease_until=?,attempt=attempt+1,updated_at=? WHERE id=? AND ((state='pending' AND fires_at<=?) OR (state='retry_wait' AND next_attempt_at<=?)) RETURNING id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,updated_at,run_id,attempt,next_attempt_at,last_error,dispatch_token,dispatch_lease_until`, token, until.UTC(), now.UTC(), id, now.UTC(), now.UTC()) } func (s *SQLiteCallbackStore) ReclaimExpired(c context.Context, id, expectedToken, token string, now, until time.Time) (CallbackInfo, bool, error) { - return s.claimReturning(c, id, token, `UPDATE delayed_callbacks SET dispatch_token=?,dispatch_lease_until=?,attempt=attempt+1,updated_at=? WHERE id=? AND state='dispatching_fenced' AND dispatch_token=? AND (dispatch_lease_until IS NULL OR dispatch_lease_until<=?) RETURNING id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,run_id,attempt,next_attempt_at,last_error,dispatch_token,dispatch_lease_until`, token, until.UTC(), now.UTC(), id, expectedToken, now.UTC()) + return s.claimReturning(c, id, token, `UPDATE delayed_callbacks SET dispatch_token=?,dispatch_lease_until=?,attempt=attempt+1,updated_at=? WHERE id=? AND state='dispatching_fenced' AND dispatch_token=? AND (dispatch_lease_until IS NULL OR dispatch_lease_until<=?) RETURNING id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,updated_at,run_id,attempt,next_attempt_at,last_error,dispatch_token,dispatch_lease_until`, token, until.UTC(), now.UTC(), id, expectedToken, now.UTC()) } // claimReturning makes claiming and reading the owner one SQLite statement. @@ -366,7 +369,7 @@ func (s *SQLiteCallbackStore) ReleaseLease(c context.Context, id, token string, } func (s *SQLiteCallbackStore) RecoverExpiredLease(c context.Context, id, expectedToken string, now time.Time) (CallbackInfo, bool, error) { - got, err := scanCallback(s.db.QueryRowContext(c, `UPDATE delayed_callbacks SET state='retry_wait',next_attempt_at=?,dispatch_token='',dispatch_lease_until=NULL,updated_at=? WHERE id=? AND state='dispatching_fenced' AND dispatch_token=? AND (dispatch_lease_until IS NULL OR dispatch_lease_until<=?) RETURNING id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,run_id,attempt,next_attempt_at,last_error,dispatch_token,dispatch_lease_until`, now.UTC(), now.UTC(), id, expectedToken, now.UTC())) + got, err := scanCallback(s.db.QueryRowContext(c, `UPDATE delayed_callbacks SET state='retry_wait',next_attempt_at=?,dispatch_token='',dispatch_lease_until=NULL,updated_at=? WHERE id=? AND state='dispatching_fenced' AND dispatch_token=? AND (dispatch_lease_until IS NULL OR dispatch_lease_until<=?) RETURNING id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,updated_at,run_id,attempt,next_attempt_at,last_error,dispatch_token,dispatch_lease_until`, now.UTC(), now.UTC(), id, expectedToken, now.UTC())) if errors.Is(err, sql.ErrNoRows) { current, getErr := s.Get(c, id) return current, false, getErr @@ -434,14 +437,14 @@ func (s *SQLiteCallbackStore) Update(c context.Context, i CallbackInfo) error { return nil } func (s *SQLiteCallbackStore) Get(c context.Context, id string) (CallbackInfo, error) { - return scanCallback(s.db.QueryRowContext(c, `SELECT id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,run_id,attempt,next_attempt_at,last_error,dispatch_token,dispatch_lease_until FROM delayed_callbacks WHERE id=?`, id)) + return scanCallback(s.db.QueryRowContext(c, `SELECT id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,updated_at,run_id,attempt,next_attempt_at,last_error,dispatch_token,dispatch_lease_until FROM delayed_callbacks WHERE id=?`, id)) } func (s *SQLiteCallbackStore) ListPending(c context.Context) ([]CallbackInfo, error) { return s.list(c, `WHERE state='pending' ORDER BY fires_at,id`) } func (s *SQLiteCallbackStore) list(c context.Context, clause string) ([]CallbackInfo, error) { - rs, e := s.db.QueryContext(c, `SELECT id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,run_id,attempt,next_attempt_at,last_error,dispatch_token,dispatch_lease_until FROM delayed_callbacks `+clause) + rs, e := s.db.QueryContext(c, `SELECT id,tenant_id,agent_id,conversation_id,prompt,model,provider_name,allow_fallback,fallback_providers,delay,fires_at,state,created_at,updated_at,run_id,attempt,next_attempt_at,last_error,dispatch_token,dispatch_lease_until FROM delayed_callbacks `+clause) if e != nil { return nil, e } @@ -472,7 +475,7 @@ func scanCallback(r callbackScanner) (CallbackInfo, error) { var i CallbackInfo var next, lease sql.NullTime var fallbackProviders string - e := r.Scan(&i.ID, &i.TenantID, &i.AgentID, &i.ConversationID, &i.Prompt, &i.Model, &i.ProviderName, &i.AllowFallback, &fallbackProviders, &i.Delay, &i.FiresAt, &i.State, &i.CreatedAt, &i.RunID, &i.Attempt, &next, &i.LastError, &i.DispatchToken, &lease) + e := r.Scan(&i.ID, &i.TenantID, &i.AgentID, &i.ConversationID, &i.Prompt, &i.Model, &i.ProviderName, &i.AllowFallback, &fallbackProviders, &i.Delay, &i.FiresAt, &i.State, &i.CreatedAt, &i.UpdatedAt, &i.RunID, &i.Attempt, &next, &i.LastError, &i.DispatchToken, &lease) if e != nil { return CallbackInfo{}, e } diff --git a/internal/harness/tools/delayed_callback_store_test.go b/internal/harness/tools/delayed_callback_store_test.go index e4fa2868..61995787 100644 --- a/internal/harness/tools/delayed_callback_store_test.go +++ b/internal/harness/tools/delayed_callback_store_test.go @@ -211,7 +211,7 @@ func TestCallbackSQLiteStoreRoundTripAndScope(t *testing.T) { if err := store.Migrate(ctx); err != nil { t.Fatal(err) } - want := CallbackInfo{ID: "cb-1", ConversationID: "conv", TenantID: "tenant", AgentID: "agent", Prompt: "hello", Delay: "5s", State: CallbackStatePending, FiresAt: time.Now().Add(time.Minute), CreatedAt: time.Now(), Model: "fixture-model", ProviderName: "missing-primary", AllowFallback: true, FallbackProviders: []string{"secondary", "tertiary"}} + want := CallbackInfo{ID: "cb-1", ConversationID: "conv", TenantID: "tenant", AgentID: "agent", Prompt: "hello", Delay: "5s", State: CallbackStatePending, FiresAt: time.Now().Add(time.Minute), CreatedAt: time.Now(), UpdatedAt: time.Now().Add(-time.Hour), Model: "fixture-model", ProviderName: "missing-primary", AllowFallback: true, FallbackProviders: []string{"secondary", "tertiary"}} if err := store.Create(ctx, want); err != nil { t.Fatal(err) } @@ -222,10 +222,20 @@ func TestCallbackSQLiteStoreRoundTripAndScope(t *testing.T) { if got.TenantID != want.TenantID || got.AgentID != want.AgentID || got.ConversationID != want.ConversationID || got.Prompt != want.Prompt || got.Model != want.Model || got.ProviderName != want.ProviderName || got.AllowFallback != want.AllowFallback || !slices.Equal(got.FallbackProviders, want.FallbackProviders) { t.Fatalf("round trip = %#v", got) } + if !got.UpdatedAt.Equal(want.UpdatedAt) { + t.Fatalf("updated_at = %v, want %v", got.UpdatedAt, want.UpdatedAt) + } pending, err := store.ListPending(ctx) if err != nil || len(pending) != 1 || pending[0].ID != want.ID { t.Fatalf("pending = %#v, err=%v", pending, err) } + canceled, err := store.CancelPending(ctx, want.ID) + if err != nil { + t.Fatal(err) + } + if canceled.State != CallbackStateCanceled || !canceled.UpdatedAt.After(want.UpdatedAt) { + t.Fatalf("cancel transition = %#v, want newer updated_at", canceled) + } } func TestCallbackSQLiteStoreClaimFencesDuplicateAndStaleToken(t *testing.T) { diff --git a/internal/harness/tools/delayed_callback_test.go b/internal/harness/tools/delayed_callback_test.go index dcef8532..de5763b2 100644 --- a/internal/harness/tools/delayed_callback_test.go +++ b/internal/harness/tools/delayed_callback_test.go @@ -296,7 +296,7 @@ func TestCallbackManagerListAll(t *testing.T) { } }) - t.Run("excludes fired and canceled callbacks", func(t *testing.T) { + t.Run("legacy conversation lists exclude fired and canceled callbacks", func(t *testing.T) { starter := &mockRunStarter{} mgr := NewCallbackManager(starter) defer mgr.Shutdown() @@ -319,7 +319,7 @@ func TestCallbackManagerListAll(t *testing.T) { } mgr.fire(fired.ID) - callbacks := mgr.ListAll() + callbacks := mgr.List("conv-1") if len(callbacks) != 1 { t.Fatalf("expected only the pending callback, got %d: %+v", len(callbacks), callbacks) } @@ -329,6 +329,73 @@ func TestCallbackManagerListAll(t *testing.T) { }) } +func TestCallbackManagerLegacyTerminalsRetainUpdateTimeAndListAll(t *testing.T) { + starter := &mockRunStarter{} + mgr := NewCallbackManager(starter) + clock := time.Date(2026, time.August, 4, 12, 0, 0, 0, time.UTC) + mgr.now = func() time.Time { return clock } + + canceled, err := mgr.Set(setReq("conv-canceled", time.Hour, "cancel me")) + if err != nil { + t.Fatalf("Set canceled: %v", err) + } + clock = clock.Add(time.Second) + if _, err := mgr.Cancel(canceled.ID); err != nil { + t.Fatalf("Cancel: %v", err) + } + + fired, err := mgr.Set(setReq("conv-fired", time.Hour, "fire me")) + if err != nil { + t.Fatalf("Set fired: %v", err) + } + mgr.mu.Lock() + mgr.callbacks[fired.ID].timer.Stop() + mgr.mu.Unlock() + clock = clock.Add(time.Second) + mgr.fire(fired.ID) + + shuttingDown, err := mgr.Set(setReq("conv-shutdown", time.Hour, "shutdown me")) + if err != nil { + t.Fatalf("Set shutdown: %v", err) + } + clock = clock.Add(time.Second) + mgr.Shutdown() + + all, err := mgr.ListAllCallbacks(context.Background()) + if err != nil { + t.Fatalf("ListAllCallbacks: %v", err) + } + byID := make(map[string]CallbackInfo, len(all)) + for _, info := range all { + byID[info.ID] = info + } + for _, want := range []struct { + info CallbackInfo + state CallbackState + }{ + {canceled, CallbackStateCanceled}, + {fired, CallbackStateFired}, + {shuttingDown, CallbackStateCanceled}, + } { + got, ok := byID[want.info.ID] + if !ok { + t.Fatalf("ListAllCallbacks missing terminal callback %s", want.info.ID) + } + if got.State != want.state || !got.UpdatedAt.After(want.info.UpdatedAt) { + t.Fatalf("terminal callback %s = %+v, want state %s with updated_at after %s", want.info.ID, got, want.state, want.info.UpdatedAt) + } + } + if got := mgr.List("conv-canceled"); len(got) != 0 { + t.Fatalf("legacy List returned canceled callback: %+v", got) + } + if got := mgr.List("conv-fired"); len(got) != 0 { + t.Fatalf("legacy List returned fired callback: %+v", got) + } + if got := mgr.List("conv-shutdown"); len(got) != 0 { + t.Fatalf("legacy List returned shutdown-canceled callback: %+v", got) + } +} + func TestCallbackManagerFire(t *testing.T) { t.Run("fire calls StartRun", func(t *testing.T) { starter := &mockRunStarter{} diff --git a/internal/server/http.go b/internal/server/http.go index 39d37360..8017dfe4 100644 --- a/internal/server/http.go +++ b/internal/server/http.go @@ -45,6 +45,7 @@ type CronClient interface { GetJob(ctx context.Context, id string) (tools.CronJob, error) UpdateJob(ctx context.Context, id string, req tools.CronUpdateJobRequest) (tools.CronJob, error) DeleteJob(ctx context.Context, id string) error + DeleteJobCAS(ctx context.Context, id string, expectedUpdatedAt time.Time) error ListExecutions(ctx context.Context, jobID string, limit, offset int) ([]tools.CronExecution, error) Health(ctx context.Context) error } diff --git a/internal/server/http_cron.go b/internal/server/http_cron.go index 0f1c5fa3..e40feeee 100644 --- a/internal/server/http_cron.go +++ b/internal/server/http_cron.go @@ -4,9 +4,11 @@ import ( "context" "encoding/json" "errors" + "io" "net/http" "strconv" "strings" + "time" "go-agent-harness/internal/harness/tools" "go-agent-harness/internal/store" @@ -232,7 +234,17 @@ func (s *Server) handleCronDeleteJob(w http.ResponseWriter, r *http.Request, id writeCronJobError(w, err) return } - if err := s.cronClient.DeleteJob(r.Context(), id); err != nil { + expectedUpdatedAt, err := cronActionExpectedUpdatedAt(r) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid_json", err.Error()) + return + } + if expectedUpdatedAt != nil { + err = s.cronClient.DeleteJobCAS(r.Context(), id, *expectedUpdatedAt) + } else { + err = s.cronClient.DeleteJob(r.Context(), id) + } + if err != nil { writeCronJobError(w, err) return } @@ -245,13 +257,23 @@ func (s *Server) handleCronPauseJob(w http.ResponseWriter, r *http.Request, id s writeMethodNotAllowed(w, http.MethodPost) return } - if _, err := s.cronJobForTenant(r.Context(), id); err != nil { + job, err := s.cronJobForTenant(r.Context(), id) + if err != nil { writeCronJobError(w, err) return } + if job.Status != "active" { + writeCronJobError(w, tools.ErrCronJobConflict) + return + } + expectedUpdatedAt, err := cronActionExpectedUpdatedAt(r) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid_json", err.Error()) + return + } paused := "paused" - job, err := s.cronClient.UpdateJob(r.Context(), id, tools.CronUpdateJobRequest{ - Status: &paused, + job, err = s.cronClient.UpdateJob(r.Context(), id, tools.CronUpdateJobRequest{ + Status: &paused, ExpectedUpdatedAt: expectedUpdatedAt, }) if err != nil { writeCronJobError(w, err) @@ -270,13 +292,23 @@ func (s *Server) handleCronResumeJob(w http.ResponseWriter, r *http.Request, id writeMethodNotAllowed(w, http.MethodPost) return } - if _, err := s.cronJobForTenant(r.Context(), id); err != nil { + job, err := s.cronJobForTenant(r.Context(), id) + if err != nil { writeCronJobError(w, err) return } + if job.Status != "paused" { + writeCronJobError(w, tools.ErrCronJobConflict) + return + } + expectedUpdatedAt, err := cronActionExpectedUpdatedAt(r) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid_json", err.Error()) + return + } active := "active" - job, err := s.cronClient.UpdateJob(r.Context(), id, tools.CronUpdateJobRequest{ - Status: &active, + job, err = s.cronClient.UpdateJob(r.Context(), id, tools.CronUpdateJobRequest{ + Status: &active, ExpectedUpdatedAt: expectedUpdatedAt, }) if err != nil { writeCronJobError(w, err) @@ -318,9 +350,29 @@ func cronJobVisibleToTenant(job tools.CronJob, tenantID string) bool { } func writeCronJobError(w http.ResponseWriter, err error) { + if errors.Is(err, tools.ErrCronJobConflict) { + writeError(w, http.StatusConflict, "conflict", "cron job changed or cannot perform that action") + return + } if errors.Is(err, tools.ErrCronJobNotFound) { writeError(w, http.StatusNotFound, "not_found", "job not found") return } writeError(w, http.StatusInternalServerError, "internal_error", err.Error()) } + +// cronActionExpectedUpdatedAt accepts an omitted/empty body for older clients, +// while letting current Activity rows submit their observed version as a CAS +// fence. Action endpoints deliberately ignore every other field. +func cronActionExpectedUpdatedAt(r *http.Request) (*time.Time, error) { + var request struct { + ExpectedUpdatedAt *time.Time `json:"expected_updated_at"` + } + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + if errors.Is(err, io.EOF) { + return nil, nil + } + return nil, err + } + return request.ExpectedUpdatedAt, nil +} diff --git a/internal/server/http_cron_test.go b/internal/server/http_cron_test.go index a6e05556..d7b6ddd0 100644 --- a/internal/server/http_cron_test.go +++ b/internal/server/http_cron_test.go @@ -6,6 +6,7 @@ import ( "crypto/rand" "encoding/base64" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -59,17 +60,18 @@ func (m *mockCronClient) CreateJob(_ context.Context, req tools.CronCreateJobReq } now := time.Now().UTC() job := tools.CronJob{ - ID: m.nextID(), - TenantID: req.TenantID, - Name: req.Name, - Schedule: req.Schedule, - ExecType: req.ExecType, - ExecConfig: req.ExecConfig, - Status: "active", - TimeoutSec: req.TimeoutSec, - Tags: req.Tags, - CreatedAt: now, - UpdatedAt: now, + ID: m.nextID(), + TenantID: req.TenantID, + ConversationID: req.ConversationID, + Name: req.Name, + Schedule: req.Schedule, + ExecType: req.ExecType, + ExecConfig: req.ExecConfig, + Status: "active", + TimeoutSec: req.TimeoutSec, + Tags: req.Tags, + CreatedAt: now, + UpdatedAt: now, } m.jobs[job.ID] = job return job, nil @@ -111,6 +113,9 @@ func (m *mockCronClient) UpdateJob(_ context.Context, id string, req tools.CronU if !ok { return tools.CronJob{}, tools.ErrCronJobNotFound } + if req.ExpectedUpdatedAt != nil && !j.UpdatedAt.Equal(*req.ExpectedUpdatedAt) { + return tools.CronJob{}, tools.ErrCronJobConflict + } if req.Status != nil { j.Status = *req.Status } @@ -144,6 +149,23 @@ func (m *mockCronClient) DeleteJob(_ context.Context, id string) error { return nil } +func (m *mockCronClient) DeleteJobCAS(_ context.Context, id string, expectedUpdatedAt time.Time) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.fail { + return fmt.Errorf("mock error") + } + job, ok := m.jobs[id] + if !ok { + return tools.ErrCronJobNotFound + } + if !job.UpdatedAt.Equal(expectedUpdatedAt) { + return tools.ErrCronJobConflict + } + delete(m.jobs, id) + return nil +} + func (m *mockCronClient) ListExecutions(_ context.Context, jobID string, limit, offset int) ([]tools.CronExecution, error) { m.mu.Lock() defer m.mu.Unlock() @@ -777,6 +799,138 @@ func TestCronResumeJob_Returns200(t *testing.T) { } } +// The Activity page can render a cron row just before another client changes +// it. Optional expected_updated_at is a CAS fence: stale actions return 409 +// and must not mutate the current job. +func TestCronLifecycleActions_OptionalExpectedUpdatedAtPreventsStaleMutation(t *testing.T) { + mock := newMockCronClient() + job, err := mock.CreateJob(context.Background(), tools.CronCreateJobRequest{ + Name: "watch deployment", Schedule: "*/5 * * * *", + }) + if err != nil { + t.Fatalf("seed cron job: %v", err) + } + ts := cronTestServer(t, mock) + stale := job.UpdatedAt.Add(-time.Second).Format(time.RFC3339Nano) + current := job.UpdatedAt.Format(time.RFC3339Nano) + + res, _ := doCronJSON(t, http.MethodPost, ts.URL+"/v1/cron/jobs/"+job.ID+"/pause", "", `{"expected_updated_at":"`+stale+`"}`) + if res.StatusCode != http.StatusConflict { + t.Fatalf("stale pause status = %d, want 409", res.StatusCode) + } + got, _ := mock.GetJob(context.Background(), job.ID) + if got.Status != "active" { + t.Fatalf("stale pause mutated status = %q, want active", got.Status) + } + + res, _ = doCronJSON(t, http.MethodPost, ts.URL+"/v1/cron/jobs/"+job.ID+"/pause", "", `{"expected_updated_at":"`+current+`"}`) + if res.StatusCode != http.StatusOK { + t.Fatalf("current pause status = %d, want 200", res.StatusCode) + } + got, _ = mock.GetJob(context.Background(), job.ID) + if got.Status != "paused" { + t.Fatalf("current pause status = %q, want paused", got.Status) + } + + res, _ = doCronJSON(t, http.MethodPost, ts.URL+"/v1/cron/jobs/"+job.ID+"/resume", "", `{"expected_updated_at":"`+current+`"}`) + if res.StatusCode != http.StatusConflict { + t.Fatalf("stale resume status = %d, want 409", res.StatusCode) + } + if got, _ = mock.GetJob(context.Background(), job.ID); got.Status != "paused" { + t.Fatalf("stale resume mutated status = %q, want paused", got.Status) + } + + current = got.UpdatedAt.Format(time.RFC3339Nano) + res, _ = doCronJSON(t, http.MethodPost, ts.URL+"/v1/cron/jobs/"+job.ID+"/resume", "", `{"expected_updated_at":"`+current+`"}`) + if res.StatusCode != http.StatusOK { + t.Fatalf("current resume status = %d, want 200", res.StatusCode) + } + got, _ = mock.GetJob(context.Background(), job.ID) + if got.Status != "active" { + t.Fatalf("current resume status = %q, want active", got.Status) + } + + res, _ = doCronJSON(t, http.MethodDelete, ts.URL+"/v1/cron/jobs/"+job.ID, "", `{"expected_updated_at":"`+current+`"}`) + if res.StatusCode != http.StatusConflict { + t.Fatalf("stale delete status = %d, want 409", res.StatusCode) + } + if _, err := mock.GetJob(context.Background(), job.ID); err != nil { + t.Fatalf("stale delete removed job: %v", err) + } + + current = got.UpdatedAt.Format(time.RFC3339Nano) + res, _ = doCronJSON(t, http.MethodDelete, ts.URL+"/v1/cron/jobs/"+job.ID, "", `{"expected_updated_at":"`+current+`"}`) + if res.StatusCode != http.StatusNoContent { + t.Fatalf("current delete status = %d, want 204", res.StatusCode) + } + if _, err := mock.GetJob(context.Background(), job.ID); !errors.Is(err, tools.ErrCronJobNotFound) { + t.Fatalf("current delete error = %v, want not found", err) + } +} + +// Task versions are opaque server strings. A task listing with nanosecond +// precision must provide the exact token callers need for the action CAS +// fence; a client-side truncated timestamp is stale even in the same second. +func TestCronLifecycleActions_TaskVersionPreservesNanoseconds(t *testing.T) { + mock := newMockCronClient() + job, err := mock.CreateJob(context.Background(), tools.CronCreateJobRequest{ + Name: "watch deployment", Schedule: "*/5 * * * *", + }) + if err != nil { + t.Fatalf("seed cron job: %v", err) + } + + updatedAt := time.Date(2026, time.August, 4, 12, 0, 0, 123456789, time.UTC) + mock.mu.Lock() + job = mock.jobs[job.ID] + job.UpdatedAt = updatedAt + mock.jobs[job.ID] = job + mock.mu.Unlock() + + ts := cronTestServer(t, mock) + res, body := doCronJSON(t, http.MethodGet, ts.URL+"/v1/tasks", "", "") + if res.StatusCode != http.StatusOK { + t.Fatalf("list tasks status = %d, want 200: %s", res.StatusCode, string(body)) + } + var listed struct { + Tasks []struct { + ID string `json:"id"` + UpdatedAt string `json:"updated_at"` + } `json:"tasks"` + } + if err := json.Unmarshal(body, &listed); err != nil { + t.Fatalf("decode task list: %v", err) + } + var token string + for _, task := range listed.Tasks { + if task.ID == job.ID { + token = task.UpdatedAt + break + } + } + want := updatedAt.Format(time.RFC3339Nano) + if token != want { + t.Fatalf("listed task version = %q, want %q", token, want) + } + + truncated := "2026-08-04T12:00:00.123Z" + res, _ = doCronJSON(t, http.MethodPost, ts.URL+"/v1/cron/jobs/"+job.ID+"/pause", "", `{"expected_updated_at":"`+truncated+`"}`) + if res.StatusCode != http.StatusConflict { + t.Fatalf("truncated pause status = %d, want 409", res.StatusCode) + } + if got, _ := mock.GetJob(context.Background(), job.ID); got.Status != "active" { + t.Fatalf("truncated pause mutated status = %q, want active", got.Status) + } + + res, _ = doCronJSON(t, http.MethodPost, ts.URL+"/v1/cron/jobs/"+job.ID+"/pause", "", `{"expected_updated_at":"`+token+`"}`) + if res.StatusCode != http.StatusOK { + t.Fatalf("exact pause status = %d, want 200", res.StatusCode) + } + if got, _ := mock.GetJob(context.Background(), job.ID); got.Status != "paused" { + t.Fatalf("exact pause status = %q, want paused", got.Status) + } +} + // TestCronEndpoints_Return501WhenNotConfigured verifies all cron endpoints return 501 when cronClient is nil. func TestCronEndpoints_Return501WhenNotConfigured(t *testing.T) { t.Parallel() diff --git a/internal/server/http_tasks.go b/internal/server/http_tasks.go index 0ee2d09d..6d4bedc8 100644 --- a/internal/server/http_tasks.go +++ b/internal/server/http_tasks.go @@ -50,17 +50,23 @@ type CallbackCanceler interface { // background work — a managed subagent, a cron job, or a pending delayed // callback — with the fields the /tasks panel needs to render a row. type Task struct { - ID string `json:"id"` - Type string `json:"type"` - Status string `json:"status"` - Label string `json:"label"` - StartedAt time.Time `json:"started_at"` - AgeSeconds int64 `json:"age_seconds"` - Actions []string `json:"actions"` - RunID string `json:"run_id,omitempty"` - Attempt int `json:"attempt,omitempty"` - NextAttemptAt time.Time `json:"next_attempt_at,omitzero"` - LastError string `json:"last_error,omitempty"` + ID string `json:"id"` + Type string `json:"type"` + Status string `json:"status"` + Label string `json:"label"` + StartedAt time.Time `json:"started_at"` + AgeSeconds int64 `json:"age_seconds"` + Actions []string `json:"actions"` + ConversationID string `json:"conversation_id,omitempty"` + NextRunAt *time.Time `json:"next_run_at,omitempty"` + LastRunAt *time.Time `json:"last_run_at,omitempty"` + FiresAt *time.Time `json:"fires_at,omitempty"` + LastExecutionStatus string `json:"last_execution_status,omitempty"` + RunID string `json:"run_id,omitempty"` + Attempt int `json:"attempt,omitempty"` + NextAttemptAt time.Time `json:"next_attempt_at,omitzero"` + LastError string `json:"last_error,omitempty"` + UpdatedAt *time.Time `json:"updated_at,omitempty"` } // handleTasks serves GET /v1/tasks: a union of every daemon-reachable piece of @@ -99,7 +105,21 @@ func (s *Server) handleTasks(w http.ResponseWriter, r *http.Request) { return } for _, job := range filterCronJobsByTenant(jobs, TenantIDFromContext(r.Context())) { - tasks = append(tasks, taskFromCronJob(job, now)) + // The task list is a lifecycle projection, so request at most the + // newest execution per job. A failed history lookup is material: a + // partially populated row could falsely imply there was no failure. + executions, err := s.cronClient.ListExecutions(r.Context(), job.ID, 1, 0) + if err != nil { + writeError(w, http.StatusInternalServerError, "internal_error", err.Error()) + return + } + var latest *tools.CronExecution + for i := range executions { + if latest == nil || executions[i].StartedAt.After(latest.StartedAt) { + latest = &executions[i] + } + } + tasks = append(tasks, taskFromCronJob(job, latest, now)) } } @@ -171,7 +191,7 @@ func taskFromSubagent(item subagents.Subagent, now time.Time) Task { // taskFromCronJob maps a cron job onto the unified DTO. Active jobs can be // paused, paused jobs resumed; every job can be deleted. -func taskFromCronJob(job tools.CronJob, now time.Time) Task { +func taskFromCronJob(job tools.CronJob, latest *tools.CronExecution, now time.Time) Task { actions := []string{TaskActionDelete} switch job.Status { case "active": @@ -179,15 +199,25 @@ func taskFromCronJob(job tools.CronJob, now time.Time) Task { case "paused": actions = []string{TaskActionResume, TaskActionDelete} } - return Task{ - ID: job.ID, - Type: TaskTypeCron, - Status: job.Status, - Label: job.Name, - StartedAt: job.CreatedAt, - AgeSeconds: taskAgeSeconds(job.CreatedAt, now), - Actions: actions, - } + task := Task{ + ID: job.ID, + Type: TaskTypeCron, + Status: job.Status, + Label: job.Name, + StartedAt: job.CreatedAt, + AgeSeconds: taskAgeSeconds(job.CreatedAt, now), + Actions: actions, + ConversationID: job.ConversationID, + NextRunAt: timePointer(job.NextRunAt), + LastRunAt: timePointer(job.LastRunAt), + UpdatedAt: timePointer(job.UpdatedAt), + } + if latest != nil { + task.LastExecutionStatus = latest.Status + task.RunID = latest.RunID + task.LastError = latest.Error + } + return task } // taskFromCallback maps every durable delayed-callback lifecycle state onto @@ -199,18 +229,29 @@ func taskFromCallback(info tools.CallbackInfo, now time.Time) Task { actions = []string{TaskActionCancel} } return Task{ - ID: info.ID, - Type: TaskTypeCallback, - Status: string(info.State), - Label: info.Prompt, - StartedAt: info.CreatedAt, - AgeSeconds: taskAgeSeconds(info.CreatedAt, now), - Actions: actions, - RunID: info.RunID, - Attempt: info.Attempt, - NextAttemptAt: info.NextAttemptAt, - LastError: tools.SafeCallbackErrorSummary(info.LastError), + ID: info.ID, + Type: TaskTypeCallback, + Status: string(info.State), + Label: info.Prompt, + StartedAt: info.CreatedAt, + AgeSeconds: taskAgeSeconds(info.CreatedAt, now), + Actions: actions, + ConversationID: info.ConversationID, + FiresAt: timePointer(info.FiresAt), + RunID: info.RunID, + Attempt: info.Attempt, + NextAttemptAt: info.NextAttemptAt, + LastError: tools.SafeCallbackErrorSummary(info.LastError), + UpdatedAt: timePointer(info.UpdatedAt), + } +} + +func timePointer(value time.Time) *time.Time { + if value.IsZero() { + return nil } + value = value.UTC() + return &value } // taskFromBashJob maps a tracked background bash job onto the unified DTO. diff --git a/internal/server/http_tasks_test.go b/internal/server/http_tasks_test.go index a12605b8..e641ceb9 100644 --- a/internal/server/http_tasks_test.go +++ b/internal/server/http_tasks_test.go @@ -182,6 +182,77 @@ func TestTasksEndpoint_UnionsAllSources(t *testing.T) { } } +// TestTasksEndpoint_ProjectsScheduledLifecycle verifies the additive task +// representation contains the timing, last execution, run linkage, and +// server-authoritative action data that native clients need to render a +// scheduled task without inventing its state from a model response. +func TestTasksEndpoint_ProjectsScheduledLifecycle(t *testing.T) { + t.Parallel() + + now := time.Date(2026, time.August, 4, 12, 0, 0, 0, time.UTC) + cronClient := newMockCronClient() + job, err := cronClient.CreateJob(context.Background(), tools.CronCreateJobRequest{ + Name: "watch deployment", Schedule: "*/5 * * * *", ConversationID: "conv-cron", + }) + if err != nil { + t.Fatalf("seed cron job: %v", err) + } + job.NextRunAt = now.Add(5 * time.Minute) + job.LastRunAt = now.Add(-5 * time.Minute) + job.UpdatedAt = now + cronClient.jobs[job.ID] = job + cronClient.executions[job.ID] = []tools.CronExecution{{ + ID: "execution-1", JobID: job.ID, StartedAt: now.Add(-5 * time.Minute), + Status: "failed", RunID: "run-cron-1", Error: "deployment unavailable", + }} + + callbacks := mockCallbackLister{callbacks: []tools.CallbackInfo{{ + ID: "callback-1", ConversationID: "conv-callback", Prompt: "say hello", + State: tools.CallbackStatePending, CreatedAt: now.Add(-time.Minute), UpdatedAt: now, + FiresAt: now.Add(time.Minute), + }}} + handler := NewWithOptions(ServerOptions{ + Runner: testRunnerForAgents(t), CronClient: cronClient, CallbackLister: callbacks, + }) + ts := httptest.NewServer(handler) + defer ts.Close() + + code, tasks := listTasks(t, ts, "") + if code != http.StatusOK { + t.Fatalf("GET /v1/tasks: status %d, want 200", code) + } + var cronTask, callbackTask *Task + for i := range tasks { + switch tasks[i].Type { + case TaskTypeCron: + cronTask = &tasks[i] + case TaskTypeCallback: + callbackTask = &tasks[i] + } + } + if cronTask == nil || callbackTask == nil { + t.Fatalf("scheduled task rows missing: %+v", tasks) + } + if cronTask.ConversationID != "conv-cron" || cronTask.NextRunAt == nil || !cronTask.NextRunAt.Equal(job.NextRunAt) || cronTask.LastRunAt == nil || !cronTask.LastRunAt.Equal(job.LastRunAt) || cronTask.UpdatedAt == nil || !cronTask.UpdatedAt.Equal(job.UpdatedAt) { + t.Errorf("cron lifecycle task = %+v, want timing, update, and conversation fields", cronTask) + } + if cronTask.LastExecutionStatus != "failed" || cronTask.RunID != "run-cron-1" || cronTask.LastError != "deployment unavailable" { + t.Errorf("cron latest execution = %+v, want failed run linkage and safe error", cronTask) + } + if got, want := cronTask.Actions, []string{TaskActionPause, TaskActionDelete}; !reflect.DeepEqual(got, want) { + t.Errorf("cron actions = %v, want %v", got, want) + } + if callbackTask.ConversationID != "conv-callback" || callbackTask.FiresAt == nil || !callbackTask.FiresAt.Equal(now.Add(time.Minute)) { + t.Errorf("callback lifecycle task = %+v, want conversation and fires_at", callbackTask) + } + if callbackTask.UpdatedAt == nil || !callbackTask.UpdatedAt.Equal(now) { + t.Errorf("callback updated_at = %v, want %v", callbackTask.UpdatedAt, now) + } + if got, want := callbackTask.Actions, []string{TaskActionCancel}; !reflect.DeepEqual(got, want) { + t.Errorf("callback actions = %v, want %v", got, want) + } +} + // TestTasksEndpoint_SkipsUnconfiguredSources verifies the union degrades // gracefully: with only a cron client configured, only cron entries appear. func TestTasksEndpoint_SkipsUnconfiguredSources(t *testing.T) { @@ -769,7 +840,7 @@ func TestJobOutputEndpoint_CrossTenant(t *testing.T) { } // TestCallbackCancelEndpoint verifies POST /v1/callbacks/{id}/cancel cancels a -// pending delayed callback. +// pending delayed callback and retains its terminal lifecycle row in /v1/tasks. func TestCallbackCancelEndpoint(t *testing.T) { t.Parallel() @@ -787,8 +858,22 @@ func TestCallbackCancelEndpoint(t *testing.T) { if code != http.StatusOK { t.Fatalf("POST /v1/callbacks/%s/cancel: status %d, body %s; want 200", info.ID, code, body) } - if got := len(mgr.ListAll()); got != 0 { - t.Fatalf("ListAll after cancel has %d pending callbacks, want 0", got) + code, tasks := listTasks(t, ts, "") + if code != http.StatusOK { + t.Fatalf("GET /v1/tasks: status %d, want 200", code) + } + if len(tasks) != 1 { + t.Fatalf("GET /v1/tasks returned %d rows, want canceled callback", len(tasks)) + } + task := tasks[0] + if task.ID != info.ID || task.Type != TaskTypeCallback || task.Status != string(tools.CallbackStateCanceled) { + t.Fatalf("canceled callback task = %+v", task) + } + if len(task.Actions) != 0 { + t.Fatalf("canceled callback actions = %v, want none", task.Actions) + } + if task.UpdatedAt == nil || task.UpdatedAt.IsZero() { + t.Fatalf("canceled callback updated_at = %v, want non-zero", task.UpdatedAt) } } diff --git a/macapp/Sources/GoCodeUI/ActivityView.swift b/macapp/Sources/GoCodeUI/ActivityView.swift index 0b01af75..5101f468 100644 --- a/macapp/Sources/GoCodeUI/ActivityView.swift +++ b/macapp/Sources/GoCodeUI/ActivityView.swift @@ -5,6 +5,7 @@ import SwiftUI /// `/dashboard` combined, since on macOS they are one "what's happening" view. struct ActivityView: View { @Bindable var project: ProjectSession + @Binding var section: Section var body: some View { ScrollView { @@ -37,7 +38,7 @@ struct ActivityView: View { LoadingPlaceholder() } else { ForEach(project.tasks) { task in - TaskRow(task: task) + TaskRow(project: project, section: $section, task: task) } } } @@ -79,33 +80,148 @@ struct ActivityView: View { } private struct TaskRow: View { + @Bindable var project: ProjectSession + @Binding var section: Section let task: TaskInfo + @State private var confirmDelete = false + @State private var actionInFlight: TaskAction? var body: some View { - HStack(spacing: Spacing.standard) { - Image(systemName: icon).foregroundStyle(.tint) - VStack(alignment: .leading, spacing: Spacing.tight) { - Text(task.label).font(Typography.body).lineLimit(1) - MetadataRow { - Text(task.type) - Text(task.status) - if let age = task.ageSeconds { Text("\(age)s") } + VStack(alignment: .leading, spacing: Spacing.small) { + HStack(spacing: Spacing.standard) { + Image(systemName: icon).foregroundStyle(.tint) + VStack(alignment: .leading, spacing: Spacing.tight) { + Text(task.label).font(Typography.body).lineLimit(1) + MetadataRow { + Text(task.type.rawValue) + Text(task.status.rawValue) + if let age = task.ageSeconds { Text("\(age)s") } + } } + Spacer() + taskActions } - Spacer() + ScheduledTaskLifecycle(task: task) + } + .accessibilityElement(children: .contain) + .accessibilityLabel(TaskLifecycleText.accessibilityLabel(for: task)) + .alert("Delete scheduled task?", isPresented: $confirmDelete) { + Button("Delete", role: .destructive) { perform(.delete) } + Button("Cancel", role: .cancel) {} + } message: { + Text("This permanently removes \(task.label).") + } + } + + @ViewBuilder + private var taskActions: some View { + if task.conversationID?.isEmpty == false { + Button("Open linked run", systemImage: "bubble.left.and.text.bubble.right") { + Task { @MainActor in + guard await project.openScheduledTask(task) else { return } + section = .chat + } + } + .buttonStyle(.borderless) + .disabled(actionInFlight != nil) + .accessibilityLabel(linkedRunAccessibilityLabel) + } + if task.actions?.contains(.pause) == true { + actionButton(.pause, title: "Pause", systemImage: "pause.fill") + } + if task.actions?.contains(.resume) == true { + actionButton(.resume, title: "Resume", systemImage: "play.fill") + } + if task.actions?.contains(.cancel) == true && task.type == .callback { + actionButton(.cancel, title: "Cancel callback", systemImage: "xmark") + } + if task.actions?.contains(.delete) == true && task.type == .cron { + Button("Delete", systemImage: "trash", role: .destructive) { confirmDelete = true } + .buttonStyle(.borderless) + .disabled(actionInFlight != nil) + .accessibilityLabel("Delete cron \(task.label)") + } + } + + private var linkedRunAccessibilityLabel: String { + if let runID = task.runID, !runID.isEmpty { + return "Open linked run \(runID) for \(task.label)" + } + return "Open linked conversation for \(task.label)" + } + + private func actionButton(_ action: TaskAction, title: String, systemImage: String) -> some View + { + Button(title, systemImage: systemImage) { perform(action) } + .buttonStyle(.borderless) + .disabled(actionInFlight != nil) + .accessibilityLabel("\(title) \(task.label)") + } + + private func perform(_ action: TaskAction) { + actionInFlight = action + Task { + await project.performTaskAction(action, for: task) + actionInFlight = nil } } private var icon: String { switch task.type { - case "subagent": return "person.2" - case "cron": return "calendar" - case "callback": return "arrow.uturn.left" + case .subagent: return "person.2" + case .cron: return "calendar" + case .callback: return "arrow.uturn.left" default: return "gearshape.2" } } } +private struct ScheduledTaskLifecycle: View { + let task: TaskInfo + + var body: some View { + let lines = TaskLifecycleText.lines(for: task) + if !lines.isEmpty { + VStack(alignment: .leading, spacing: Spacing.tight) { + ForEach(lines, id: \.self) { line in + Text(line).font(Typography.detail).foregroundStyle(Theme.foregroundTertiary) + } + } + .accessibilityLabel(TaskLifecycleText.accessibilityLabel(for: task)) + } + } +} + +enum TaskLifecycleText { + static func lines(for task: TaskInfo) -> [String] { + var lines: [String] = [] + if let nextRunAt = task.nextRunAt { + lines.append("Next: \(nextRunAt.formatted(date: .omitted, time: .shortened))") + } + if let firesAt = task.firesAt { + lines.append("Due: \(firesAt.formatted(date: .omitted, time: .shortened))") + } + if let lastRunAt = task.lastRunAt { + lines.append("Last: \(lastRunAt.formatted(date: .omitted, time: .shortened))") + } + if let status = task.lastExecutionStatus { lines.append("Last result: \(status.rawValue)") } + if let runID = task.runID { lines.append("Run: \(runID)") } + if let attempt = task.attempt { lines.append("Attempt: \(attempt)") } + if let nextAttemptAt = task.nextAttemptAt { + lines.append("Retry: \(nextAttemptAt.formatted(date: .omitted, time: .shortened))") + } + if let lastError = task.lastError, !lastError.isEmpty { + lines.append("Error: \(lastError)") + } + return lines + } + + static func accessibilityLabel(for task: TaskInfo) -> String { + ([task.type.rawValue, task.label, task.status.rawValue] + lines(for: task)).joined( + separator: ", ") + } +} + private struct RunRow: View { let run: RunSummaryInfo diff --git a/macapp/Sources/GoCodeUI/AppShell.swift b/macapp/Sources/GoCodeUI/AppShell.swift index 865ce6fc..ab08e179 100644 --- a/macapp/Sources/GoCodeUI/AppShell.swift +++ b/macapp/Sources/GoCodeUI/AppShell.swift @@ -148,7 +148,7 @@ private struct ProjectView: View { ChatView(project: project, run: run) } case .activity: - ActivityView(project: project) + ActivityView(project: project, section: $section) case .sessions: SessionsView(project: project, section: $section) case .checkpoints: diff --git a/macapp/Sources/GoCodeUI/EnvironmentInspector.swift b/macapp/Sources/GoCodeUI/EnvironmentInspector.swift index 0c3343c5..3563cf05 100644 --- a/macapp/Sources/GoCodeUI/EnvironmentInspector.swift +++ b/macapp/Sources/GoCodeUI/EnvironmentInspector.swift @@ -10,8 +10,8 @@ struct EnvironmentInspector: View { let activities: [ToolActivity] @Binding var selected: ToolActivity? - private var subagents: [TaskInfo] { project.tasks.filter { $0.type == "subagent" } } - private var backgroundTasks: [TaskInfo] { project.tasks.filter { $0.type != "subagent" } } + private var subagents: [TaskInfo] { project.tasks.filter { $0.type == .subagent } } + private var backgroundTasks: [TaskInfo] { project.tasks.filter { $0.type != .subagent } } /// Distinguishes "nothing here" from "not fetched yet", so the card does /// not assert an absence it cannot know. private var hasPendingCollectionLoad: Bool { @@ -136,7 +136,8 @@ private struct TaskSummary: View { HStack(spacing: Spacing.small) { Text(task.label).font(Typography.caption).lineLimit(1) Spacer(minLength: Spacing.none) - Text(task.status).font(Typography.detail).foregroundStyle(Theme.foregroundTertiary) + Text(task.status.rawValue).font(Typography.detail).foregroundStyle( + Theme.foregroundTertiary) } } } diff --git a/macapp/Sources/GoCodeUI/ProjectSession.swift b/macapp/Sources/GoCodeUI/ProjectSession.swift index e3446b8e..0ffa95a1 100644 --- a/macapp/Sources/GoCodeUI/ProjectSession.swift +++ b/macapp/Sources/GoCodeUI/ProjectSession.swift @@ -311,6 +311,56 @@ public final class ProjectSession { } } + /// Executes only a server-advertised scheduled-task action, then reloads + /// the authoritative task union even when the action fails. This avoids a + /// stale Activity row claiming an action succeeded after a concurrent + /// scheduler transition, restart, or permission change. + public func performTaskAction(_ action: TaskAction, for task: TaskInfo) async { + guard let client, task.actions?.contains(action) == true else { return } + do { + switch (task.type, action) { + case (.cron, .pause): + try await client.pauseCron(id: task.id, expectedUpdatedAt: task.updatedAtVersion) + case (.cron, .resume): + try await client.resumeCron(id: task.id, expectedUpdatedAt: task.updatedAtVersion) + case (.cron, .delete): + try await client.deleteCron(id: task.id, expectedUpdatedAt: task.updatedAtVersion) + case (.callback, .cancel): + try await client.cancelCallback(id: task.id) + default: + // Generic task actions either have a different existing + // control path or are an unknown additive server action. + return + } + } catch { + statusMessage = error.localizedDescription + } + await refreshActivity() + } + + /// Opens the actual conversation behind a scheduled task. A linked run is + /// only made the active chat control target after its own non-terminal + /// event passes through RunSession's reducer; terminal history opens as a + /// transcript without inventing Stop/approve/steer authority. + @discardableResult + public func openScheduledTask(_ task: TaskInfo) async -> Bool { + guard let client, let conversationID = task.conversationID, !conversationID.isEmpty else { + return false + } + do { + let messages = try await client.messages(conversationID: conversationID) + run?.load(messages: messages, conversationID: conversationID) + await refreshRewindPoints() + if let runID = task.runID, !runID.isEmpty { + await run?.attachLinkedActiveRun(runID: runID) + } + return true + } catch { + statusMessage = error.localizedDescription + return false + } + } + /// Rehydrates the selected conversation from durable messages when Chat /// becomes visible. Conversation SSE remains the low-latency path; this is /// the durability safety net for a completed scheduled run that happened diff --git a/macapp/Sources/GoCodeUI/RunSession.swift b/macapp/Sources/GoCodeUI/RunSession.swift index 8d8621c7..b1977cf6 100644 --- a/macapp/Sources/GoCodeUI/RunSession.swift +++ b/macapp/Sources/GoCodeUI/RunSession.swift @@ -320,6 +320,24 @@ public final class RunSession { trackConversationStream(conversationID) } + /// Attaches an Activity-linked run only when the daemon supplies active + /// run evidence. A terminal replay remains transcript history and must + /// never create a fake live control target. + func attachLinkedActiveRun(runID: String) async { + guard !runID.isEmpty, !terminalRunIDs.contains(runID) else { return } + do { + for try await event in client.events(runID: runID) { + guard event.runID == runID else { continue } + _ = await apply(event, runID: runID) + if currentRunID == runID || event.type.isTerminal { return } + } + } catch is CancellationError { + return + } catch { + connectionError = error.localizedDescription + } + } + public func recallPreviousPrompt() { guard let last = promptHistory.last else { return } draft = last diff --git a/macapp/Sources/HarnessKit/ClientTasks.swift b/macapp/Sources/HarnessKit/ClientTasks.swift index 1ff82e91..9bde2e1a 100644 --- a/macapp/Sources/HarnessKit/ClientTasks.swift +++ b/macapp/Sources/HarnessKit/ClientTasks.swift @@ -1,23 +1,185 @@ import Foundation +/// Forward-compatible task kind from the server's background-work union. +public enum TaskType: Sendable, Hashable, Decodable { + case subagent, cron, callback, bashJob + case unknown(String) + + public init(from decoder: Decoder) throws { + self.init(rawValue: try decoder.singleValueContainer().decode(String.self)) + } + + public init(rawValue: String) { + switch rawValue { + case "subagent": self = .subagent + case "cron": self = .cron + case "callback": self = .callback + case "bash_job": self = .bashJob + default: self = .unknown(rawValue) + } + } + + public var rawValue: String { + switch self { + case .subagent: return "subagent" + case .cron: return "cron" + case .callback: return "callback" + case .bashJob: return "bash_job" + case .unknown(let value): return value + } + } +} + +/// Forward-compatible lifecycle state. Task-specific states deliberately stay +/// distinct from `RunStatus`: a callback can be pending while no run exists. +public enum TaskStatus: Sendable, Hashable, Decodable { + case active, paused, pending, retryWait, dispatching, started, fired, canceled, running, + completed, failed, skipped, exited + case unknown(String) + + public init(from decoder: Decoder) throws { + self.init(rawValue: try decoder.singleValueContainer().decode(String.self)) + } + + public init(rawValue: String) { + switch rawValue { + case "active": self = .active + case "paused": self = .paused + case "pending": self = .pending + case "retry_wait": self = .retryWait + case "dispatching": self = .dispatching + case "started": self = .started + case "fired": self = .fired + case "canceled", "cancelled": self = .canceled + case "running": self = .running + case "completed": self = .completed + case "failed": self = .failed + case "skipped": self = .skipped + case "exited": self = .exited + default: self = .unknown(rawValue) + } + } + + public var rawValue: String { + switch self { + case .active: return "active" + case .paused: return "paused" + case .pending: return "pending" + case .retryWait: return "retry_wait" + case .dispatching: return "dispatching" + case .started: return "started" + case .fired: return "fired" + case .canceled: return "canceled" + case .running: return "running" + case .completed: return "completed" + case .failed: return "failed" + case .skipped: return "skipped" + case .exited: return "exited" + case .unknown(let value): return value + } + } +} + +/// An action advertised by the server for one task row. Unknown actions are +/// retained for diagnostics but never rendered as an executable control. +public enum TaskAction: Sendable, Hashable, Decodable { + case cancel, delete, pause, resume + case unknown(String) + + public init(from decoder: Decoder) throws { + self.init(rawValue: try decoder.singleValueContainer().decode(String.self)) + } + + public init(rawValue: String) { + switch rawValue { + case "cancel": self = .cancel + case "delete": self = .delete + case "pause": self = .pause + case "resume": self = .resume + default: self = .unknown(rawValue) + } + } + + public var rawValue: String { + switch self { + case .cancel: return "cancel" + case .delete: return "delete" + case .pause: return "pause" + case .resume: return "resume" + case .unknown(let value): return value + } + } +} + /// A piece of background work the daemon knows about: a subagent, cron job, or -/// pending delayed callback. +/// delayed callback. Scheduled lifecycle fields are optional so clients remain +/// compatible with older additive task payloads. public struct TaskInfo: Sendable, Decodable, Identifiable, Hashable { public let id: String - public let type: String - public let status: String + public let type: TaskType + public let status: TaskStatus public let label: String public let startedAt: Date? public let ageSeconds: Int? - public let actions: [String]? + public let actions: [TaskAction]? + public let conversationID: String? + public let nextRunAt: Date? + public let lastRunAt: Date? + public let firesAt: Date? + public let lastExecutionStatus: TaskStatus? + public let runID: String? + public let attempt: Int? + public let nextAttemptAt: Date? + public let lastError: String? + /// Opaque server version token used for cron action compare-and-swap. + /// Keep this wire value as text: decoding and reformatting it as `Date` + /// can lose RFC3339 nanoseconds and make a fresh row look stale. + public let updatedAtVersion: String? + + public init( + id: String, type: TaskType, status: TaskStatus, label: String, + startedAt: Date? = nil, ageSeconds: Int? = nil, actions: [TaskAction]? = nil, + conversationID: String? = nil, nextRunAt: Date? = nil, lastRunAt: Date? = nil, + firesAt: Date? = nil, lastExecutionStatus: TaskStatus? = nil, + runID: String? = nil, attempt: Int? = nil, nextAttemptAt: Date? = nil, + lastError: String? = nil, updatedAtVersion: String? = nil + ) { + self.id = id + self.type = type + self.status = status + self.label = label + self.startedAt = startedAt + self.ageSeconds = ageSeconds + self.actions = actions + self.conversationID = conversationID + self.nextRunAt = nextRunAt + self.lastRunAt = lastRunAt + self.firesAt = firesAt + self.lastExecutionStatus = lastExecutionStatus + self.runID = runID + self.attempt = attempt + self.nextAttemptAt = nextAttemptAt + self.lastError = lastError + self.updatedAtVersion = updatedAtVersion + } enum CodingKeys: String, CodingKey { case id, type, status, label, actions case startedAt = "started_at" case ageSeconds = "age_seconds" + case conversationID = "conversation_id" + case nextRunAt = "next_run_at" + case lastRunAt = "last_run_at" + case firesAt = "fires_at" + case lastExecutionStatus = "last_execution_status" + case runID = "run_id" + case attempt + case nextAttemptAt = "next_attempt_at" + case lastError = "last_error" + case updatedAtVersion = "updated_at" } - public var isCancellable: Bool { actions?.contains("cancel") ?? false } + public var isCancellable: Bool { actions?.contains(.cancel) ?? false } } public struct TodoItem: Sendable, Decodable, Identifiable, Hashable { @@ -62,6 +224,40 @@ extension HarnessClient { return response.tasks ?? [] } + /// Each task control calls a type-specific, server-authorized endpoint. + /// The caller must follow the request with `tasks()` rather than mutate a + /// local row optimistically because allowed actions can have gone stale. + public func pauseCron(id: String, expectedUpdatedAt: String? = nil) async throws { + try await cronAction( + .post, "/v1/cron/jobs/\(id)/pause", expectedUpdatedAt: expectedUpdatedAt) + } + + public func resumeCron(id: String, expectedUpdatedAt: String? = nil) async throws { + try await cronAction( + .post, "/v1/cron/jobs/\(id)/resume", expectedUpdatedAt: expectedUpdatedAt) + } + + public func deleteCron(id: String, expectedUpdatedAt: String? = nil) async throws { + try await cronAction(.delete, "/v1/cron/jobs/\(id)", expectedUpdatedAt: expectedUpdatedAt) + } + + public func cancelCallback(id: String) async throws { + try await sendVoid(.post, "/v1/callbacks/\(id)/cancel") + } + + /// Preserves an empty request body for older additive task payloads, while + /// current task rows send their observed version as a CAS fence. + private func cronAction(_ method: Method, _ path: String, expectedUpdatedAt: String?) + async throws + { + if let expectedUpdatedAt { + try await sendVoid( + method, path, body: TaskActionVersion(expectedUpdatedAt: expectedUpdatedAt)) + } else { + try await sendVoid(method, path) + } + } + public func todos(runID: String) async throws -> [TodoItem] { struct Response: Decodable { let todos: [TodoItem]? } let response: Response = try await get("/v1/runs/\(runID)/todos") @@ -92,3 +288,9 @@ extension HarnessClient { } } } + +private struct TaskActionVersion: Encodable { + let expectedUpdatedAt: String + + enum CodingKeys: String, CodingKey { case expectedUpdatedAt = "expected_updated_at" } +} diff --git a/macapp/Tests/GoCodeUITests/ProjectSessionActivityTests.swift b/macapp/Tests/GoCodeUITests/ProjectSessionActivityTests.swift index 1997ddd0..7825e2f0 100644 --- a/macapp/Tests/GoCodeUITests/ProjectSessionActivityTests.swift +++ b/macapp/Tests/GoCodeUITests/ProjectSessionActivityTests.swift @@ -82,6 +82,13 @@ struct ProjectSessionActivityTests { } } + private final class Paths: @unchecked Sendable { + private let lock = NSLock() + private var values: [String] = [] + func append(_ value: String) { lock.withLock { values.append(value) } } + var snapshot: [String] { lock.withLock { values } } + } + @Test("a genuine 501 stays nil with no error, but a real failure surfaces via statusMessage") func distinguishesTransportErrorFromNoStoreSignal() async throws { let project = makeProject() @@ -215,4 +222,119 @@ struct ProjectSessionActivityTests { }, "persisted scheduled reply did not appear when Chat re-entered") } + + @Test("a stale task action surfaces the server error and reconciles its authoritative row") + func staleTaskActionReconcilesAuthoritativeRow() async throws { + let project = makeProject() + let active = Data( + #"{"tasks":[{"id":"cron-1","type":"cron","status":"active","label":"watch","started_at":"2026-08-04T12:00:00Z","age_seconds":0,"actions":["pause","delete"]}]}"# + .utf8) + let paused = Data( + #"{"tasks":[{"id":"cron-1","type":"cron","status":"paused","label":"watch","started_at":"2026-08-04T12:00:00Z","age_seconds":0,"actions":["resume","delete"]}]}"# + .utf8) + ActivityStubProtocol.set { request in + switch request.url?.path { + case "/v1/tasks": return .init(body: active) + default: return .init(body: Data("{}".utf8)) + } + } + await project.start() + await project.refreshActivity() + let task = try #require(project.tasks.first) + let paths = Paths() + ActivityStubProtocol.set { request in + let path = request.url?.path ?? "" + paths.append(path) + switch path { + case "/v1/cron/jobs/cron-1/pause": + return .init( + status: 409, + body: Data(#"{"error":{"code":"conflict","message":"already paused"}}"#.utf8)) + case "/v1/tasks": return .init(body: paused) + default: return .init(body: Data("{}".utf8)) + } + } + + await project.performTaskAction(.pause, for: task) + + #expect(paths.snapshot.contains("/v1/cron/jobs/cron-1/pause")) + #expect(paths.snapshot.contains("/v1/tasks")) + #expect(project.tasks.first?.status == .paused) + #expect(project.statusMessage == "already paused") + } + + @Test("scheduled lifecycle labels expose result, timing, run link, and safe error") + func lifecycleLabelsAreAccessible() throws { + let task = TaskInfo( + id: "cron-1", type: .cron, status: .active, label: "watch deploy", + lastExecutionStatus: .failed, runID: "run-1", lastError: "deployment unavailable") + let label = TaskLifecycleText.accessibilityLabel(for: task) + #expect(label.contains("cron")) + #expect(label.contains("watch deploy")) + #expect(label.contains("Last result: failed")) + #expect(label.contains("Run: run-1")) + #expect(label.contains("Error: deployment unavailable")) + } + + @Test("opening a linked scheduled task loads its chat and attaches only active run evidence") + func opensLinkedScheduledTaskWithActiveRun() async throws { + let project = makeProject() + ActivityStubProtocol.set { request in + switch request.url?.path { + case "/v1/conversations/conv-linked/messages": + return .init( + body: Data( + #"{"messages":[{"role":"user","content":"watch deploy","step":0}]}"#.utf8)) + case "/v1/runs/run-linked/events": + return .init( + body: Data( + "event: run.started\ndata: {\"id\":\"run-linked:0\",\"run_id\":\"run-linked\",\"type\":\"run.started\",\"payload\":{}}\n\n" + .utf8)) + default: + return .init(body: Data("{}".utf8)) + } + } + await project.start() + let opened = await project.openScheduledTask( + TaskInfo( + id: "cron-1", type: .cron, status: .active, label: "watch deploy", + conversationID: "conv-linked", runID: "run-linked")) + #expect(opened) + #expect(project.run?.conversationID == "conv-linked") + #expect(project.run?.currentRunID == "run-linked") + } + + @Test("terminal or missing task links never manufacture live run controls") + func terminalAndMissingTaskLinksDoNotCreateLiveControls() async throws { + let project = makeProject() + ActivityStubProtocol.set { request in + switch request.url?.path { + case "/v1/conversations/conv-terminal/messages": + return .init( + body: Data( + #"{"messages":[{"role":"assistant","content":"done","step":0}]}"#.utf8)) + case "/v1/runs/run-terminal/events": + return .init( + body: Data( + "event: run.completed\ndata: {\"id\":\"run-terminal:1\",\"run_id\":\"run-terminal\",\"type\":\"run.completed\",\"payload\":{}}\n\n" + .utf8)) + default: + return .init(body: Data("{}".utf8)) + } + } + await project.start() + let terminalOpened = await project.openScheduledTask( + TaskInfo( + id: "cron-1", type: .cron, status: .active, label: "watch deploy", + conversationID: "conv-terminal", runID: "run-terminal")) + #expect(terminalOpened) + #expect(project.run?.conversationID == "conv-terminal") + #expect(project.run?.currentRunID == nil) + + let missingOpened = await project.openScheduledTask( + TaskInfo( + id: "orphan", type: .cron, status: .active, label: "orphan", runID: "run-orphan")) + #expect(!missingOpened) + #expect(project.run?.currentRunID == nil) + } } diff --git a/macapp/Tests/HarnessKitTests/ClientTasksTests.swift b/macapp/Tests/HarnessKitTests/ClientTasksTests.swift new file mode 100644 index 00000000..dbd293f9 --- /dev/null +++ b/macapp/Tests/HarnessKitTests/ClientTasksTests.swift @@ -0,0 +1,148 @@ +import Foundation +import Testing + +@testable import HarnessKit + +/// Isolated transport stub for this suite. `StubURLProtocol` is intentionally +/// process-global for the older HarnessClient suite, so reusing it here would +/// let independent Swift Testing suites overwrite each other's handler. +private final class TaskStubURLProtocol: URLProtocol, @unchecked Sendable { + struct Response: Sendable { + var status: Int = 200 + var body = Data() + } + + nonisolated(unsafe) private static var handler: (@Sendable (URLRequest) -> Response)? + nonisolated(unsafe) private static var recorded: [URLRequest] = [] + nonisolated(unsafe) private static var recordedBodies: [Data?] = [] + private static let lock = NSLock() + + static func set(_ handler: @escaping @Sendable (URLRequest) -> Response) { + lock.withLock { + self.handler = handler + self.recorded = [] + self.recordedBodies = [] + } + } + + static var requests: [URLRequest] { lock.withLock { recorded } } + static var bodies: [Data?] { lock.withLock { recordedBodies } } + + override class func canInit(with request: URLRequest) -> Bool { true } + override class func canonicalRequest(for request: URLRequest) -> URLRequest { request } + + override func startLoading() { + let request = self.request + Self.lock.withLock { + Self.recorded.append(request) + Self.recordedBodies.append(Self.bodyData(request)) + } + let response = Self.lock.withLock { Self.handler }?(request) ?? Response() + let http = HTTPURLResponse( + url: request.url!, statusCode: response.status, httpVersion: "HTTP/1.1", + headerFields: ["Content-Type": "application/json"])! + client?.urlProtocol(self, didReceive: http, cacheStoragePolicy: .notAllowed) + client?.urlProtocol(self, didLoad: response.body) + client?.urlProtocolDidFinishLoading(self) + } + + override func stopLoading() {} + + private static func bodyData(_ request: URLRequest) -> Data? { + if let body = request.httpBody { return body } + guard let stream = request.httpBodyStream else { return nil } + stream.open() + defer { stream.close() } + var data = Data() + var buffer = [UInt8](repeating: 0, count: 4096) + while stream.hasBytesAvailable { + let read = stream.read(&buffer, maxLength: buffer.count) + if read <= 0 { break } + data.append(buffer, count: read) + } + return data + } +} + +@Suite("HarnessClient task lifecycle", .serialized) +struct ClientTasksTests { + + private func makeClient() -> HarnessClient { + let configuration = URLSessionConfiguration.ephemeral + configuration.protocolClasses = [TaskStubURLProtocol.self] + return HarnessClient( + baseURL: URL(string: "http://127.0.0.1:8899")!, + session: URLSession(configuration: configuration)) + } + + @Test("decodes additive lifecycle fields and preserves unknown task values") + func decodesLifecycleAndUnknownValues() async throws { + TaskStubURLProtocol.set { request in + #expect(request.httpMethod == "GET") + #expect(request.url?.path == "/v1/tasks") + return .init( + body: Data( + #""" + {"tasks":[{ + "id":"cron-1","type":"cron","status":"active","label":"watch deploy", + "started_at":"2026-08-04T12:00:00Z","age_seconds":1, + "actions":["pause","delete","future_action"], + "conversation_id":"conv-1","next_run_at":"2026-08-04T12:05:00Z", + "last_run_at":"2026-08-04T11:55:00Z","last_execution_status":"completed", + "run_id":"run-1","updated_at":"2026-08-04T12:00:00.123456789Z" + },{ + "id":"future-1","type":"future_kind","status":"future_state","label":"future", + "started_at":"2026-08-04T12:00:00Z","age_seconds":0,"actions":["future_action"] + }]} + """#.utf8)) + } + + let tasks = try await makeClient().tasks() + let cron = try #require(tasks.first) + #expect(cron.type == .cron) + #expect(cron.status == .active) + #expect(cron.conversationID == "conv-1") + #expect(cron.nextRunAt != nil) + #expect(cron.lastRunAt != nil) + #expect(cron.lastExecutionStatus == .completed) + #expect(cron.runID == "run-1") + #expect(cron.updatedAtVersion == "2026-08-04T12:00:00.123456789Z") + #expect(cron.actions?.contains(.pause) == true) + #expect(cron.actions?.contains(.unknown("future_action")) == true) + #expect(tasks[1].type == .unknown("future_kind")) + #expect(tasks[1].status == .unknown("future_state")) + } + + @Test("uses scoped task-control endpoints with the row version when present") + func sendsTaskControlRequests() async throws { + TaskStubURLProtocol.set { request in + #expect(request.httpMethod == "POST" || request.httpMethod == "DELETE") + return .init(status: request.httpMethod == "DELETE" ? 204 : 200, body: Data("{}".utf8)) + } + let client = makeClient() + let updatedAt = "2026-08-04T12:00:00.123456789Z" + try await client.pauseCron(id: "cron-1", expectedUpdatedAt: updatedAt) + try await client.resumeCron(id: "cron-1", expectedUpdatedAt: updatedAt) + try await client.deleteCron(id: "cron-1", expectedUpdatedAt: updatedAt) + try await client.cancelCallback(id: "callback-1") + + #expect( + TaskStubURLProtocol.requests.map(\.url?.path) == [ + "/v1/cron/jobs/cron-1/pause", "/v1/cron/jobs/cron-1/resume", + "/v1/cron/jobs/cron-1", "/v1/callbacks/callback-1/cancel", + ]) + for body in TaskStubURLProtocol.bodies.prefix(3) { + let body = try #require(body) + let payload = try #require( + JSONSerialization.jsonObject(with: body) as? [String: String]) + #expect(payload["expected_updated_at"] == updatedAt) + } + } + + @Test("preserves empty action bodies when no task version is available") + func preservesEmptyActionBodiesWithoutVersion() async throws { + TaskStubURLProtocol.set { _ in .init(body: Data("{}".utf8)) } + try await makeClient().pauseCron(id: "cron-1") + #expect(TaskStubURLProtocol.bodies == [nil]) + } +}