feat(reflection): expose worker runtime status
CI / test (push) Successful in 46s
CI / test (pull_request) Successful in 46s

This commit is contained in:
m
2026-08-23 19:38:51 +03:00
parent c81e2269e1
commit 666ec09ef0
5 changed files with 154 additions and 2 deletions
+17
View File
@@ -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)
}
}
+74
View File
@@ -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
}
+35
View File
@@ -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)
}
}