feat(agent-worker): expose recovery cycle evidence
CI / test (push) Successful in 47s
CI / test (pull_request) Successful in 46s

This commit is contained in:
m
2026-08-23 19:50:19 +03:00
parent ea90bf0fab
commit 4839f139e8
5 changed files with 161 additions and 22 deletions
+11 -1
View File
@@ -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)
}
+95 -19
View File
@@ -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)
})
}
+30
View File
@@ -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)
}
}
+2 -2
View File
@@ -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
}
@@ -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"]
}