Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7978b495d1 | ||
|
|
d73d4fc118 |
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"]
|
||||
}
|
||||
Reference in New Issue
Block a user