From 7978b495d160701bd55b092114fb50eddb8963f0 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 23 Aug 2026 19:38:51 +0300 Subject: [PATCH] feat(reflection): expose worker runtime status --- cmd/reflection-worker/main.go | 17 +++++ cmd/reflection-worker/status.go | 74 +++++++++++++++++++ cmd/reflection-worker/status_test.go | 35 +++++++++ dev/ACTIVE_TASK.json | 4 +- ...00244-reflection-worker-observability.json | 26 +++++++ 5 files changed, 154 insertions(+), 2 deletions(-) create mode 100644 cmd/reflection-worker/status.go create mode 100644 cmd/reflection-worker/status_test.go create mode 100644 dev/tasks/DEV-000244-reflection-worker-observability.json diff --git a/cmd/reflection-worker/main.go b/cmd/reflection-worker/main.go index 1898cd5..fa75273 100644 --- a/cmd/reflection-worker/main.go +++ b/cmd/reflection-worker/main.go @@ -4,6 +4,8 @@ import ( "context" "flag" "fmt" + "log" + "net/http" "os" "time" @@ -29,9 +31,21 @@ func main() { panic(err) } defer repository.Close() + status := newWorkerStatus(*workerID, *once) + listenAddr := os.Getenv("REFLECTION_LISTEN_ADDR") + if listenAddr == "" { + listenAddr = ":8092" + } + go func() { + if err := http.ListenAndServe(listenAddr, status.handler()); err != nil { + log.Printf("reflection status server stopped: %v", err) + } + }() for { now := time.Now().UnixMilli() + status.cycleStarted(time.Now()) if _, err := repository.RecoverReflectionClaims(ctx, now); err != nil { + status.cycleFailed(time.Now(), err) panic(err) } candidate, claimErr := repository.ClaimReflectionCandidate(ctx, *workerID, now, lease.Milliseconds()) @@ -39,16 +53,19 @@ func main() { // This worker only closes the durable evaluation loop. Any cognitive // proposal remains a candidate and must be admitted by a separate governed flow. if err := repository.ResolveReflectionCandidate(ctx, candidate.ID, *workerID, time.Now().UnixMilli()); err != nil { + status.cycleFailed(time.Now(), err) panic(err) } fmt.Printf("resolved reflection candidate %s\n", candidate.ID) } else if *once { fmt.Println("no reflection candidate") } + status.cycleFinished(time.Now(), map[bool]string{true: "candidate-resolved", false: "idle"}[claimErr == nil]) if *once { return } if err := repository.WaitForReflectionCandidate(ctx, *pollInterval); err != nil { + status.cycleFailed(time.Now(), err) panic(err) } } diff --git a/cmd/reflection-worker/status.go b/cmd/reflection-worker/status.go new file mode 100644 index 0000000..d65d39f --- /dev/null +++ b/cmd/reflection-worker/status.go @@ -0,0 +1,74 @@ +package main + +import ( + "encoding/json" + "net/http" + "sync" + "time" + + "git.cin.su/m/Assistant/runtime/health" +) + +type workerStatus struct { + mu sync.RWMutex + workerID string + once bool + startedAt time.Time + lastCycleAt time.Time + lastCompleted time.Time + lastResult string + lastError string +} + +func newWorkerStatus(workerID string, once bool) *workerStatus { + return &workerStatus{workerID: workerID, once: once, startedAt: time.Now().UTC()} +} + +func (s *workerStatus) cycleStarted(at time.Time) { + s.mu.Lock() + defer s.mu.Unlock() + s.lastCycleAt = at.UTC() + s.lastResult = "running" + s.lastError = "" +} + +func (s *workerStatus) cycleFinished(at time.Time, result string) { + s.mu.Lock() + defer s.mu.Unlock() + s.lastCompleted = at.UTC() + s.lastResult = result + s.lastError = "" +} + +func (s *workerStatus) cycleFailed(at time.Time, err error) { + s.mu.Lock() + defer s.mu.Unlock() + s.lastCompleted = at.UTC() + s.lastResult = "failed" + s.lastError = err.Error() +} + +func (s *workerStatus) handler() http.Handler { + mux := http.NewServeMux() + mux.Handle("/healthz", health.Handler{Runtime: "reflection-worker", SemanticGeneration: 1, ArchitectureGeneration: 1}) + mux.HandleFunc("/status", func(w http.ResponseWriter, _ *http.Request) { + s.mu.RLock() + response := struct { + Runtime string `json:"runtime"` + WorkerID string `json:"worker_id"` + Mode string `json:"mode"` + StartedAt time.Time `json:"started_at"` + LastCycleAt time.Time `json:"last_cycle_at,omitempty"` + LastCompleted time.Time `json:"last_completed_at,omitempty"` + LastResult string `json:"last_result"` + LastError string `json:"last_error,omitempty"` + }{Runtime: "reflection-worker", WorkerID: s.workerID, Mode: "durable-reflection-loop", StartedAt: s.startedAt, LastCycleAt: s.lastCycleAt, LastCompleted: s.lastCompleted, LastResult: s.lastResult, LastError: s.lastError} + if s.once { + response.Mode = "durable-reflection-once" + } + s.mu.RUnlock() + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(response) + }) + return mux +} diff --git a/cmd/reflection-worker/status_test.go b/cmd/reflection-worker/status_test.go new file mode 100644 index 0000000..343223f --- /dev/null +++ b/cmd/reflection-worker/status_test.go @@ -0,0 +1,35 @@ +package main + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestWorkerStatusReportsCycle(t *testing.T) { + status := newWorkerStatus("test-worker", false) + status.cycleStarted(time.Date(2026, 8, 23, 10, 0, 0, 0, time.UTC)) + status.cycleFinished(time.Date(2026, 8, 23, 10, 0, 1, 0, time.UTC), "idle") + + recorder := httptest.NewRecorder() + status.handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/status", nil)) + if recorder.Code != http.StatusOK { + t.Fatalf("status code = %d, want 200", recorder.Code) + } + body := recorder.Body.String() + for _, expected := range []string{"\"runtime\":\"reflection-worker\"", "\"worker_id\":\"test-worker\"", "\"last_result\":\"idle\""} { + if !strings.Contains(body, expected) { + t.Fatalf("status body missing %q: %s", expected, body) + } + } +} + +func TestWorkerHealthIsAvailable(t *testing.T) { + recorder := httptest.NewRecorder() + newWorkerStatus("test-worker", true).handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/healthz", nil)) + if recorder.Code != http.StatusOK { + t.Fatalf("health status code = %d, want 200", recorder.Code) + } +} diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 4c5b143..337ce96 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000243-qualification-selection-evidence.json", - "task_id": "DEV-000243", + "task_file": "dev/tasks/DEV-000244-reflection-worker-observability.json", + "task_id": "DEV-000244", "task_revision": 1 } diff --git a/dev/tasks/DEV-000244-reflection-worker-observability.json b/dev/tasks/DEV-000244-reflection-worker-observability.json new file mode 100644 index 0000000..5349bff --- /dev/null +++ b/dev/tasks/DEV-000244-reflection-worker-observability.json @@ -0,0 +1,26 @@ +{ + "task_id": "DEV-000244", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "c81e2269e157c1a41b567d79d93bbbe49dd4b134", + "risk_tier": "E", + "intent": "Expose bounded health and status evidence for the durable reflection worker without changing reflection semantics or authority.", + "requirement_ids": ["learning reflection loop", "runtime observability", "fast operational verification"], + "allowed_change_scope": { + "modules": ["epistemics"], + "paths": ["cmd/reflection-worker/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000244-reflection-worker-observability.json"] + }, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "database/**", "LLM-local/**", "modules/work/secrets/**", "contracts/**"]}, + "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 reflection repository port", "runtime health conventions"], "may_change": ["worker operational HTTP surface"]}, + "semantic_owner": "Epistemics reflection worker runtime", + "canonical_state": ["none; status is derived operational evidence"], + "state_transitions": ["cycle start/complete/failure are reported without changing candidate lifecycle rules"], + "side_effects_and_privileges": ["read-only health/status endpoint; existing repository writes unchanged"], + "persistence_and_migrations": ["none"], + "failure_retry_restart": ["status resets on process restart; durable candidate state remains authoritative"], + "assumptions": ["internal production network is sufficient for worker status checks"], + "acceptance_scenarios": ["health endpoint returns 200", "status exposes loop mode and latest cycle", "reflection loop still resolves candidates"], + "required_tests": ["go test ./cmd/reflection-worker", "go vet ./cmd/reflection-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"] +} -- 2.54.0