diff --git a/cmd/agent-worker/main.go b/cmd/agent-worker/main.go index 1e118f6..37024dd 100644 --- a/cmd/agent-worker/main.go +++ b/cmd/agent-worker/main.go @@ -56,13 +56,15 @@ func main() { executor = newTaskEvaluationExecutor(envOr("COGNITIVE_WORKER_URL", "http://cognitive-worker:8083"), envOr("WORK_GRAPH_RUNTIME_URL", "http://work-graph-runtime:8088"), envOr("TASK_VERIFIER_URL", "http://task-verifier:8091"), &http.Client{Timeout: *leaseTTL}) fmt.Fprintln(os.Stderr, "protected agent execution bridge URL is not configured; task evaluations use the local durable evaluator, other work remains in fallback") } + runtimeStatus := newAgentWorkerRuntimeStatus() go func() { - if serveErr := http.ListenAndServe(*statusAddr, workerStatusHandler(executor != nil && !localFallback, localFallback)); serveErr != nil { + if serveErr := http.ListenAndServe(*statusAddr, workerStatusHandler(executor != nil && !localFallback, localFallback, runtimeStatus)); serveErr != nil { fmt.Fprintln(os.Stderr, "agent worker status server:", serveErr) } }() ctx := context.Background() for { + runtimeStatus.cycleStarted(time.Now()) var result application.WorkerCycleResult var cycleErr error if executor == nil { @@ -71,11 +73,19 @@ func main() { result, cycleErr = application.ExecuteWorkerCycleWithExecutor(ctx, repository, executor, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL) } if cycleErr != nil { + runtimeStatus.cycleFailed(time.Now(), cycleErr) fmt.Fprintln(os.Stderr, cycleErr) if *once { os.Exit(1) } } + cycleResult := "idle" + if result.Completed { + cycleResult = "completed" + } else if result.RecoveredClaims > 0 { + cycleResult = "recovered" + } + runtimeStatus.cycleFinished(time.Now(), result.RecoveredClaims, result.DispatchID, cycleResult) if cycleErr == nil && (result.Completed || result.RecoveredClaims > 0) { _ = json.NewEncoder(os.Stdout).Encode(result) } diff --git a/cmd/agent-worker/status.go b/cmd/agent-worker/status.go index 89bdcb8..a8176fd 100644 --- a/cmd/agent-worker/status.go +++ b/cmd/agent-worker/status.go @@ -3,22 +3,87 @@ package main import ( "encoding/json" "net/http" + "sync" + "time" ) type workerStatusResponse struct { - Runtime string `json:"runtime"` - ExecutorConfigured bool `json:"executor_configured"` - Mode string `json:"mode"` - Ready bool `json:"ready"` - Reason string `json:"reason"` - LocalFallback bool `json:"local_fallback"` - LocalExecutor bool `json:"local_executor"` - Capabilities []string `json:"capabilities,omitempty"` + Runtime string `json:"runtime"` + ExecutorConfigured bool `json:"executor_configured"` + Mode string `json:"mode"` + Ready bool `json:"ready"` + Reason string `json:"reason"` + LocalFallback bool `json:"local_fallback"` + LocalExecutor bool `json:"local_executor"` + Capabilities []string `json:"capabilities,omitempty"` + StartedAt time.Time `json:"started_at,omitempty"` + LastCycleAt time.Time `json:"last_cycle_at,omitempty"` + LastCompletedAt time.Time `json:"last_completed_at,omitempty"` + LastRecoveredClaims int64 `json:"last_recovered_claims"` + TotalRecoveredClaims int64 `json:"total_recovered_claims"` + LastDispatchID string `json:"last_dispatch_id,omitempty"` + LastResult string `json:"last_result,omitempty"` + LastError string `json:"last_error,omitempty"` } -func workerStatusHandler(executorConfigured bool, localFallback ...bool) http.Handler { - local := len(localFallback) > 0 && localFallback[0] - status := workerStatusResponse{ +type agentWorkerRuntimeStatus struct { + mu sync.RWMutex + startedAt time.Time + lastCycleAt time.Time + lastCompletedAt time.Time + lastRecoveredClaims int64 + totalRecoveredClaims int64 + lastDispatchID string + lastResult string + lastError string +} + +func newAgentWorkerRuntimeStatus() *agentWorkerRuntimeStatus { + return &agentWorkerRuntimeStatus{startedAt: time.Now().UTC()} +} + +func (s *agentWorkerRuntimeStatus) cycleStarted(at time.Time) { + s.mu.Lock() + defer s.mu.Unlock() + s.lastCycleAt = at.UTC() + s.lastResult = "running" + s.lastError = "" +} + +func (s *agentWorkerRuntimeStatus) cycleFinished(at time.Time, recoveredClaims int64, dispatchID string, result string) { + s.mu.Lock() + defer s.mu.Unlock() + s.lastCompletedAt = at.UTC() + s.lastRecoveredClaims = recoveredClaims + s.totalRecoveredClaims += recoveredClaims + s.lastDispatchID = dispatchID + s.lastResult = result + s.lastError = "" +} + +func (s *agentWorkerRuntimeStatus) cycleFailed(at time.Time, err error) { + s.mu.Lock() + defer s.mu.Unlock() + s.lastCompletedAt = at.UTC() + s.lastResult = "failed" + s.lastError = err.Error() +} + +func workerStatusHandler(executorConfigured bool, localFallback ...interface{}) http.Handler { + local := false + var runtime *agentWorkerRuntimeStatus + for _, argument := range localFallback { + switch value := argument.(type) { + case bool: + local = value + case *agentWorkerRuntimeStatus: + runtime = value + } + } + if runtime == nil { + runtime = newAgentWorkerRuntimeStatus() + } + baseStatus := workerStatusResponse{ Runtime: "agent-worker", ExecutorConfigured: executorConfigured, Mode: "recovery-only", @@ -28,14 +93,14 @@ func workerStatusHandler(executorConfigured bool, localFallback ...bool) http.Ha LocalExecutor: local, } if local { - status.Mode = "local-executor" - status.Reason = "bounded local task-evaluation executor is active; unsupported work remains durably waiting" - status.Capabilities = []string{"task-evaluation"} + baseStatus.Mode = "local-executor" + baseStatus.Reason = "bounded local task-evaluation executor is active; unsupported work remains durably waiting" + baseStatus.Capabilities = []string{"task-evaluation"} } if executorConfigured { - status.Mode = "execution" - status.Ready = true - status.Reason = "typed agent execution bridge configured" + baseStatus.Mode = "execution" + baseStatus.Ready = true + baseStatus.Reason = "typed agent execution bridge configured" } return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet || (r.URL.Path != "/healthz" && r.URL.Path != "/readyz" && r.URL.Path != "/status") { @@ -43,9 +108,20 @@ func workerStatusHandler(executorConfigured bool, localFallback ...bool) http.Ha return } w.Header().Set("Content-Type", "application/json") - if r.URL.Path == "/readyz" && !status.Ready { + if r.URL.Path == "/readyz" && !baseStatus.Ready { w.WriteHeader(http.StatusServiceUnavailable) } - _ = json.NewEncoder(w).Encode(status) + response := baseStatus + runtime.mu.RLock() + response.StartedAt = runtime.startedAt + response.LastCycleAt = runtime.lastCycleAt + response.LastCompletedAt = runtime.lastCompletedAt + response.LastRecoveredClaims = runtime.lastRecoveredClaims + response.TotalRecoveredClaims = runtime.totalRecoveredClaims + response.LastDispatchID = runtime.lastDispatchID + response.LastResult = runtime.lastResult + response.LastError = runtime.lastError + runtime.mu.RUnlock() + _ = json.NewEncoder(w).Encode(response) }) } diff --git a/cmd/agent-worker/status_test.go b/cmd/agent-worker/status_test.go index ac1d913..84bb185 100644 --- a/cmd/agent-worker/status_test.go +++ b/cmd/agent-worker/status_test.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "testing" + "time" ) func TestLoadProtectedBridgeURL(t *testing.T) { @@ -77,3 +78,32 @@ func TestWorkerReadyzRequiresExecutor(t *testing.T) { t.Fatalf("configured readyz status=%d", response.Code) } } + +func TestWorkerStatusReportsRecoveryCycle(t *testing.T) { + runtime := newAgentWorkerRuntimeStatus() + runtime.cycleStarted(time.Date(2026, 8, 23, 10, 0, 0, 0, time.UTC)) + runtime.cycleFinished(time.Date(2026, 8, 23, 10, 0, 1, 0, time.UTC), 3, "dispatch-1", "recovered") + response := httptest.NewRecorder() + workerStatusHandler(false, true, runtime).ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/status", nil)) + var value workerStatusResponse + if err := json.NewDecoder(response.Body).Decode(&value); err != nil { + t.Fatal(err) + } + if value.LastRecoveredClaims != 3 || value.TotalRecoveredClaims != 3 || value.LastDispatchID != "dispatch-1" || value.LastResult != "recovered" { + t.Fatalf("recovery evidence missing: %+v", value) + } +} + +func TestWorkerStatusReportsCycleError(t *testing.T) { + runtime := newAgentWorkerRuntimeStatus() + runtime.cycleFailed(time.Date(2026, 8, 23, 10, 0, 1, 0, time.UTC), os.ErrClosed) + response := httptest.NewRecorder() + workerStatusHandler(false, runtime).ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/status", nil)) + var value workerStatusResponse + if err := json.NewDecoder(response.Body).Decode(&value); err != nil { + t.Fatal(err) + } + if value.LastResult != "failed" || value.LastError == "" { + t.Fatalf("cycle error missing: %+v", value) + } +} diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 337ce96..d763569 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000244-reflection-worker-observability.json", - "task_id": "DEV-000244", + "task_file": "dev/tasks/DEV-000245-agent-worker-recovery-status.json", + "task_id": "DEV-000245", "task_revision": 1 } diff --git a/dev/tasks/DEV-000245-agent-worker-recovery-status.json b/dev/tasks/DEV-000245-agent-worker-recovery-status.json new file mode 100644 index 0000000..0d126c2 --- /dev/null +++ b/dev/tasks/DEV-000245-agent-worker-recovery-status.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000245", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "ea90bf0fabea43b735dcb901a2da2733db216369", + "risk_tier": "E", + "intent": "Expose live durable agent-worker cycle and recovery evidence through the existing internal status surface without changing dispatch, lease, executor, or canonical semantics.", + "requirement_ids": ["Agent Execution runtime", "recovery observability", "fast operational verification"], + "allowed_change_scope": {"modules": ["work"], "paths": ["cmd/agent-worker/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000245-agent-worker-recovery-status.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "database/**", "deploy/**", "contracts/**", "LLM-local/**", "modules/work/secrets/**"]}, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 0}, + "contracts": {"consume": ["existing WorkerCycleResult", "existing worker status HTTP surface"], "may_change": ["derived worker status response fields"]}, + "semantic_owner": "Work agent-worker runtime observability", + "canonical_state": ["none; status is derived operational evidence"], + "state_transitions": ["cycle start -> completed/failed status only"], + "side_effects_and_privileges": ["read-only internal status endpoint; no new privilege"], + "persistence_and_migrations": ["none"], + "failure_retry_restart": ["status is process-local evidence and resets on restart; durable recovery remains repository-owned"], + "assumptions": ["existing status address is reachable from creator-web and internal diagnostics"], + "acceptance_scenarios": ["status reports cycle completion", "status reports recovered claims", "cycle error is visible without exposing secrets", "existing executor mode remains accurate"], + "required_tests": ["go test ./cmd/agent-worker", "go vet ./cmd/agent-worker", "go run ./cmd/devctl preflight -active dev/ACTIVE_TASK.json", "go run ./cmd/devctl check-diff -active dev/ACTIVE_TASK.json", "go run ./cmd/devctl check-architecture", "git diff --check", "independent", "CI", "production health smoke"], + "terminal_gap_states": ["BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_SPEC_GAP"] +}