feat(work-graph): expose runtime status #223
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
})
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"]
|
||||
}
|
||||
Reference in New Issue
Block a user