From 2353f1f3bb35c45399e17f14075ddf0640ac7004 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 23 Aug 2026 19:57:32 +0300 Subject: [PATCH] feat(work-graph): expose runtime status --- adapters/postgres/work_repository.go | 48 ++++++++++++++++ cmd/work-graph-runtime/main.go | 1 + cmd/work-graph-runtime/status.go | 45 +++++++++++++++ cmd/work-graph-runtime/status_test.go | 56 +++++++++++++++++++ dev/ACTIVE_TASK.json | 4 +- .../DEV-000246-work-graph-runtime-status.json | 23 ++++++++ 6 files changed, 175 insertions(+), 2 deletions(-) create mode 100644 cmd/work-graph-runtime/status.go create mode 100644 cmd/work-graph-runtime/status_test.go create mode 100644 dev/tasks/DEV-000246-work-graph-runtime-status.json diff --git a/adapters/postgres/work_repository.go b/adapters/postgres/work_repository.go index 82b6b59..53659ed 100644 --- a/adapters/postgres/work_repository.go +++ b/adapters/postgres/work_repository.go @@ -32,6 +32,54 @@ func OpenWork(ctx context.Context, config ContinuityConfig) (*WorkRepository, er } return &WorkRepository{pool}, nil } + +// WorkGraphRuntimeStatus is a read-only operational projection owned by Work. +// It is not canonical graph state and cannot advance any lifecycle. +type WorkGraphRuntimeStatus struct { + GraphsByState map[string]int64 `json:"graphs_by_state"` + NodesByState map[string]int64 `json:"nodes_by_state"` + QueueByState map[string]int64 `json:"queue_by_state"` + ObservedAtUnixMillis int64 `json:"observed_at_unix_millis"` +} + +func (r *WorkRepository) RuntimeStatus(ctx context.Context) (WorkGraphRuntimeStatus, error) { + status := WorkGraphRuntimeStatus{ + GraphsByState: map[string]int64{}, + NodesByState: map[string]int64{}, + QueueByState: map[string]int64{}, + ObservedAtUnixMillis: time.Now().UnixMilli(), + } + queries := []struct { + query string + destination map[string]int64 + }{ + {`SELECT state, count(*) FROM work.graphs GROUP BY state`, status.GraphsByState}, + {`SELECT state, count(*) FROM work.graph_nodes GROUP BY state`, status.NodesByState}, + {`SELECT state, count(*) FROM work.agent_dispatch_queue GROUP BY state`, status.QueueByState}, + } + for _, item := range queries { + rows, err := r.pool.Query(ctx, item.query) + if err != nil { + return WorkGraphRuntimeStatus{}, err + } + for rows.Next() { + var state string + var count int64 + if err := rows.Scan(&state, &count); err != nil { + rows.Close() + return WorkGraphRuntimeStatus{}, err + } + item.destination[state] = count + } + if err := rows.Err(); err != nil { + rows.Close() + return WorkGraphRuntimeStatus{}, err + } + rows.Close() + } + return status, nil +} + func (r *WorkRepository) Close() { r.pool.Close() } func (r *WorkRepository) Ping(ctx context.Context) error { return r.pool.Ping(ctx) } func (r *WorkRepository) Create(ctx context.Context, target work.TaskTarget) error { diff --git a/cmd/work-graph-runtime/main.go b/cmd/work-graph-runtime/main.go index 74b006a..3392f9c 100644 --- a/cmd/work-graph-runtime/main.go +++ b/cmd/work-graph-runtime/main.go @@ -35,6 +35,7 @@ func main() { mux := http.NewServeMux() mux.Handle("/healthz", health.Handler{Runtime: "work-graph-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1}) mux.Handle("/readyz", health.ReadinessHandler{Runtime: "work-graph-runtime", Probes: []health.Probe{{Name: "work-postgres", Check: func(ctx context.Context) error { return repository.Ping(ctx) }}}}) + mux.Handle("/v1/work-graphs/status", workGraphStatusHandler(repository)) mux.HandleFunc("POST /v1/work-graphs/schedule", func(w http.ResponseWriter, r *http.Request) { var request application.GraphScheduleRequest if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&request); err != nil { diff --git a/cmd/work-graph-runtime/status.go b/cmd/work-graph-runtime/status.go new file mode 100644 index 0000000..11a9ad3 --- /dev/null +++ b/cmd/work-graph-runtime/status.go @@ -0,0 +1,45 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + + "git.cin.su/m/Assistant/adapters/postgres" +) + +type workGraphStatusReader interface { + RuntimeStatus(context.Context) (postgres.WorkGraphRuntimeStatus, error) +} + +type workGraphStatusResponse struct { + Runtime string `json:"runtime"` + Status string `json:"status"` + GraphsByState map[string]int64 `json:"graphs_by_state"` + NodesByState map[string]int64 `json:"nodes_by_state"` + QueueByState map[string]int64 `json:"queue_by_state"` + ObservedAtUnixMillis int64 `json:"observed_at_unix_millis"` +} + +func workGraphStatusHandler(reader workGraphStatusReader) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || r.URL.Path != "/v1/work-graphs/status" { + http.NotFound(w, r) + return + } + status, err := reader.RuntimeStatus(r.Context()) + if err != nil { + http.Error(w, "work graph status unavailable", http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(workGraphStatusResponse{ + Runtime: "work-graph-runtime", + Status: "ready", + GraphsByState: status.GraphsByState, + NodesByState: status.NodesByState, + QueueByState: status.QueueByState, + ObservedAtUnixMillis: status.ObservedAtUnixMillis, + }) + }) +} diff --git a/cmd/work-graph-runtime/status_test.go b/cmd/work-graph-runtime/status_test.go new file mode 100644 index 0000000..c011860 --- /dev/null +++ b/cmd/work-graph-runtime/status_test.go @@ -0,0 +1,56 @@ +package main + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + + "git.cin.su/m/Assistant/adapters/postgres" +) + +type fakeWorkGraphStatusReader struct { + status postgres.WorkGraphRuntimeStatus + err error +} + +func (f fakeWorkGraphStatusReader) RuntimeStatus(context.Context) (postgres.WorkGraphRuntimeStatus, error) { + return f.status, f.err +} + +func TestWorkGraphStatusReportsDurableCounts(t *testing.T) { + response := httptest.NewRecorder() + workGraphStatusHandler(fakeWorkGraphStatusReader{status: postgres.WorkGraphRuntimeStatus{ + GraphsByState: map[string]int64{"RUNNING": 2}, + NodesByState: map[string]int64{"LEASED": 3, "SUCCEEDED": 1}, + QueueByState: map[string]int64{"QUEUED": 3}, + ObservedAtUnixMillis: 1700000000000, + }}).ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/v1/work-graphs/status", nil)) + if response.Code != http.StatusOK { + t.Fatalf("status code = %d", response.Code) + } + body := response.Body.String() + for _, expected := range []string{"\"status\":\"ready\"", "\"RUNNING\":2", "\"LEASED\":3", "\"QUEUED\":3", "1700000000000"} { + if !contains(body, expected) { + t.Fatalf("response missing %q: %s", expected, body) + } + } +} + +func TestWorkGraphStatusFailsClosedOnDatabaseError(t *testing.T) { + response := httptest.NewRecorder() + workGraphStatusHandler(fakeWorkGraphStatusReader{err: errors.New("database unavailable")}).ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/v1/work-graphs/status", nil)) + if response.Code != http.StatusServiceUnavailable { + t.Fatalf("status code = %d", response.Code) + } +} + +func contains(value, expected string) bool { + for i := 0; i+len(expected) <= len(value); i++ { + if value[i:i+len(expected)] == expected { + return true + } + } + return false +} diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index d763569..0bc3418 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000245-agent-worker-recovery-status.json", - "task_id": "DEV-000245", + "task_file": "dev/tasks/DEV-000246-work-graph-runtime-status.json", + "task_id": "DEV-000246", "task_revision": 1 } diff --git a/dev/tasks/DEV-000246-work-graph-runtime-status.json b/dev/tasks/DEV-000246-work-graph-runtime-status.json new file mode 100644 index 0000000..592046f --- /dev/null +++ b/dev/tasks/DEV-000246-work-graph-runtime-status.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000246", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "355ba7db5621a11b2c4e0827d9301ab7fdb2d419", + "risk_tier": "E", + "intent": "Expose read-only durable Work Graph runtime counts for operational verification without changing graph scheduling, fan-out/fan-in, verifier, or replan semantics.", + "requirement_ids": ["Multi-Agent Work Graph", "fan-out/fan-in observability", "fast operational verification"], + "allowed_change_scope": {"modules": ["work"], "paths": ["adapters/postgres/work_repository.go", "cmd/work-graph-runtime/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000246-work-graph-runtime-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": ["work-owned graph and dispatch queue tables", "existing runtime health conventions"], "may_change": ["internal read-only status response"]}, + "semantic_owner": "Work Graph runtime observability", + "canonical_state": ["none; counts are derived projections"], + "state_transitions": ["none"], + "side_effects_and_privileges": ["read-only aggregate queries only"], + "persistence_and_migrations": ["none"], + "failure_retry_restart": ["status fails closed with 503 when work database is unavailable"], + "assumptions": ["status is consumed by internal diagnostics and can be queried from the production network"], + "acceptance_scenarios": ["readyz remains database-backed", "status returns graph state and queue counts", "status does not mutate canonical state"], + "required_tests": ["go test ./cmd/work-graph-runtime ./adapters/postgres", "go vet ./cmd/work-graph-runtime ./adapters/postgres", "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