From 785441f2a0065f33ebb826bdf2e8c43e903b04b8 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sat, 22 Aug 2026 16:03:16 +0300 Subject: [PATCH] feat: add durable work graph runtime --- adapters/postgres/work_repository.go | 9 +- cmd/work-graph-runtime/main.go | 88 ++++++++++++ contracts/work/graph.go | 29 ++-- .../035_work_graph_dispatch_binding.sql | 8 ++ deploy/docker-compose.prod.yml | 16 +++ dev/ACTIVE_TASK.json | 4 +- dev/tasks/DEV-000146-work-graph-runtime.json | 23 ++++ modules/work/application/graph_runtime.go | 130 ++++++++++++++++++ modules/work/ports/graph.go | 14 ++ tests/agent_execution/graph_runtime_test.go | 57 ++++++++ 10 files changed, 358 insertions(+), 20 deletions(-) create mode 100644 cmd/work-graph-runtime/main.go create mode 100644 database/migrations/035_work_graph_dispatch_binding.sql create mode 100644 dev/tasks/DEV-000146-work-graph-runtime.json create mode 100644 modules/work/application/graph_runtime.go create mode 100644 modules/work/ports/graph.go create mode 100644 tests/agent_execution/graph_runtime_test.go diff --git a/adapters/postgres/work_repository.go b/adapters/postgres/work_repository.go index 008c226..47669d6 100644 --- a/adapters/postgres/work_repository.go +++ b/adapters/postgres/work_repository.go @@ -26,7 +26,8 @@ func OpenWork(ctx context.Context, config ContinuityConfig) (*WorkRepository, er } return &WorkRepository{pool}, nil } -func (r *WorkRepository) Close() { r.pool.Close() } +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 { _, err := r.pool.Exec(ctx, `INSERT INTO work.tasks (task_id,goal_id,scope_id,revision) VALUES ($1,$2,$3,$4) ON CONFLICT (task_id) DO NOTHING`, target.TaskID, target.GoalID, target.ScopeID, target.Revision) return err @@ -143,7 +144,7 @@ func (r *WorkRepository) SaveWorkGraph(ctx context.Context, graph work.WorkGraph return err } for _, node := range graph.Nodes { - if _, err := tx.Exec(ctx, `INSERT INTO work.graph_nodes(graph_id,revision,node_id,idempotency_key,state,attempts) VALUES($1,$2,$3,$4,$5,$6) ON CONFLICT(graph_id,revision,node_id) DO NOTHING`, graph.ID, graph.Revision, node.ID, node.IdempotencyKey, node.State, node.Attempts); err != nil { + if _, err := tx.Exec(ctx, `INSERT INTO work.graph_nodes(graph_id,revision,node_id,dispatch_id,idempotency_key,state,attempts) VALUES($1,$2,$3,$4,$5,$6,$7) ON CONFLICT(graph_id,revision,node_id) DO NOTHING`, graph.ID, graph.Revision, node.ID, node.DispatchID, node.IdempotencyKey, node.State, node.Attempts); err != nil { return err } } @@ -161,13 +162,13 @@ func (r *WorkRepository) LoadWorkGraph(ctx context.Context, graphID string, revi if err != nil { return work.WorkGraph{}, fmt.Errorf("load graph: %w", err) } - rows, err := r.pool.Query(ctx, `SELECT node_id,idempotency_key,state,attempts FROM work.graph_nodes WHERE graph_id=$1 AND revision=$2 ORDER BY node_id`, graphID, revision) + rows, err := r.pool.Query(ctx, `SELECT node_id,dispatch_id,idempotency_key,state,attempts FROM work.graph_nodes WHERE graph_id=$1 AND revision=$2 ORDER BY node_id`, graphID, revision) if err != nil { return work.WorkGraph{}, err } for rows.Next() { var node work.GraphNode - if err := rows.Scan(&node.ID, &node.IdempotencyKey, &node.State, &node.Attempts); err != nil { + if err := rows.Scan(&node.ID, &node.DispatchID, &node.IdempotencyKey, &node.State, &node.Attempts); err != nil { rows.Close() return work.WorkGraph{}, err } diff --git a/cmd/work-graph-runtime/main.go b/cmd/work-graph-runtime/main.go new file mode 100644 index 0000000..62ee482 --- /dev/null +++ b/cmd/work-graph-runtime/main.go @@ -0,0 +1,88 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + "net/http" + "os" + + "git.cin.su/m/Assistant/adapters/postgres" + secretcontracts "git.cin.su/m/Assistant/contracts/secrets" + "git.cin.su/m/Assistant/modules/work/application" + "git.cin.su/m/Assistant/platform/secrets" + "git.cin.su/m/Assistant/runtime/health" +) + +func main() { + ctx := context.Background() + password, err := resolveDatabasePassword() + if err != nil { + log.Fatal(err) + } + repository, err := postgres.OpenWork(ctx, postgres.ContinuityConfig{Host: os.Getenv("POSTGRES_HOST"), Port: os.Getenv("POSTGRES_PORT"), Database: os.Getenv("POSTGRES_DATABASE"), User: os.Getenv("WORK_POSTGRES_USER"), Password: password}) + if err != nil { + log.Fatal(err) + } + defer repository.Close() + 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.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 { + http.Error(w, "invalid graph schedule request", http.StatusBadRequest) + return + } + graph, ready, err := application.AdmitAndSchedule(r.Context(), repository, request) + if err != nil { + http.Error(w, err.Error(), http.StatusUnprocessableEntity) + return + } + writeJSON(w, map[string]any{"graph": graph, "scheduled_nodes": ready}) + }) + mux.HandleFunc("POST /v1/work-graphs/complete", func(w http.ResponseWriter, r *http.Request) { + var request application.GraphCompleteRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&request); err != nil { + http.Error(w, "invalid graph completion request", http.StatusBadRequest) + return + } + graph, err := application.CompleteGraphNode(r.Context(), repository, request.GraphID, request.Revision, request.NodeID, request.Succeeded) + if err != nil { + http.Error(w, err.Error(), http.StatusUnprocessableEntity) + return + } + writeJSON(w, graph) + }) + addr := os.Getenv("LISTEN_ADDR") + if addr == "" { + addr = ":8088" + } + log.Printf("work-graph-runtime listening on %s", addr) + log.Fatal(http.ListenAndServe(addr, mux)) +} + +func writeJSON(w http.ResponseWriter, value any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(value) +} + +func resolveDatabasePassword() (string, error) { + if _, ok := os.LookupEnv("WORK_POSTGRES_PASSWORD"); ok { + return "", fmt.Errorf("WORK_POSTGRES_PASSWORD direct environment secret is forbidden") + } + path, ref := os.Getenv("WORK_POSTGRES_PASSWORD_FILE"), os.Getenv("WORK_POSTGRES_CREDENTIAL_REF") + if path == "" || ref == "" { + return "", fmt.Errorf("WORK_POSTGRES_PASSWORD_FILE and WORK_POSTGRES_CREDENTIAL_REF are required") + } + root := os.Getenv("SECRET_BACKEND_ROOT") + if root == "" { + root = "/run/secrets" + } + resolver, err := secrets.NewFileResolver(root) + if err != nil { + return "", err + } + return resolver.ResolveFile(secretcontracts.SecretFileRef{CredentialRef: secretcontracts.CredentialRef(ref), Purpose: secretcontracts.SecretPurposeDatabasePassword, Consumer: secretcontracts.RuntimeConsumerWorkWorker, Path: path}) +} diff --git a/contracts/work/graph.go b/contracts/work/graph.go index f9c83b9..f6b58bf 100644 --- a/contracts/work/graph.go +++ b/contracts/work/graph.go @@ -26,29 +26,30 @@ const ( ) type GraphBudget struct { - MaxParallel int - MaxAttempts int + MaxParallel int `json:"max_parallel"` + MaxAttempts int `json:"max_attempts"` } type GraphNode struct { - ID GraphNodeID - IdempotencyKey string - State GraphNodeState - Attempts int + ID GraphNodeID `json:"id"` + DispatchID string `json:"dispatch_id"` + IdempotencyKey string `json:"idempotency_key"` + State GraphNodeState `json:"state"` + Attempts int `json:"attempts"` } type GraphEdge struct { - From GraphNodeID - To GraphNodeID + From GraphNodeID `json:"from"` + To GraphNodeID `json:"to"` } type WorkGraph struct { - ID string - Revision int64 - State GraphState - Nodes []GraphNode - Edges []GraphEdge - Budget GraphBudget + ID string `json:"id"` + Revision int64 `json:"revision"` + State GraphState `json:"state"` + Nodes []GraphNode `json:"nodes"` + Edges []GraphEdge `json:"edges"` + Budget GraphBudget `json:"budget"` } type ResultCandidate struct { diff --git a/database/migrations/035_work_graph_dispatch_binding.sql b/database/migrations/035_work_graph_dispatch_binding.sql new file mode 100644 index 0000000..3de5158 --- /dev/null +++ b/database/migrations/035_work_graph_dispatch_binding.sql @@ -0,0 +1,8 @@ +BEGIN; + +SET ROLE work_role; + +ALTER TABLE work.graph_nodes ADD COLUMN IF NOT EXISTS dispatch_id TEXT NOT NULL DEFAULT ''; + +RESET ROLE; +COMMIT; diff --git a/deploy/docker-compose.prod.yml b/deploy/docker-compose.prod.yml index 2be2d27..89e9334 100644 --- a/deploy/docker-compose.prod.yml +++ b/deploy/docker-compose.prod.yml @@ -118,6 +118,22 @@ services: volumes: - assistant_core_secrets:/run/secrets:ro + work-graph-runtime: + build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: work-graph-runtime, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } } + restart: unless-stopped + environment: + LISTEN_ADDR: :8088 + POSTGRES_HOST: postgres + POSTGRES_PORT: "5432" + POSTGRES_DATABASE: ${POSTGRES_DB:-assistant} + WORK_POSTGRES_USER: work_role + WORK_POSTGRES_CREDENTIAL_REF: credential://domovoy/work/db-password/v1 + WORK_POSTGRES_PASSWORD_FILE: /run/secrets/work_db_password + SECRET_BACKEND_ROOT: /run/secrets + depends_on: { postgres: { condition: service_healthy } } + volumes: + - assistant_core_secrets:/run/secrets:ro + reflection-worker: build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: reflection-worker, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } } restart: unless-stopped diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index d371789..e8529cb 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000145-agent-execution-bridge-runtime.json", - "task_id": "DEV-000145", + "task_file": "dev/tasks/DEV-000146-work-graph-runtime.json", + "task_id": "DEV-000146", "task_revision": 1 } diff --git a/dev/tasks/DEV-000146-work-graph-runtime.json b/dev/tasks/DEV-000146-work-graph-runtime.json new file mode 100644 index 0000000..875e606 --- /dev/null +++ b/dev/tasks/DEV-000146-work-graph-runtime.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000146", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "16d714238e6e77a105731311eaa54f3f486d3431", + "risk_tier": "E", + "intent": "Provide a durable work-graph runtime with bounded fan-out scheduling, dispatch queue binding, and revisioned fan-in completion.", + "requirement_ids": ["WP-400", "Multi-Agent Work Graph"], + "allowed_change_scope": {"modules": ["work", "work-graph-runtime"], "paths": ["contracts/work/graph.go", "modules/work/application/**", "modules/work/ports/**", "adapters/postgres/work_repository.go", "database/migrations/035_work_graph_dispatch_binding.sql", "cmd/work-graph-runtime/**", "deploy/docker-compose.prod.yml", "tests/agent_execution/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000146-work-graph-runtime.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", ".gitea/**", "contracts/authority/**", "modules/identity/**"]}, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 1, "new_third_party_dependencies": 0, "new_database_owner_roles": 1}, + "contracts": {"consume": ["WorkGraph", "AgentDispatchIntent", "AgentDispatchQueueItem"], "may_change": ["GraphNode dispatch binding serialization"]}, + "semantic_owner": "work", + "canonical_state": ["work graph revision, graph nodes, graph edges, dispatch intent and queue"], + "state_transitions": ["DRAFT -> ADMITTED -> RUNNING -> PARTIAL/COMPLETED or FAILED", "BLOCKED_PREREQUISITE -> LEASED -> SUCCEEDED/FAILED", "each transition writes a new graph revision"], + "side_effects_and_privileges": ["work-role local graph and dispatch writes only"], + "persistence_and_migrations": ["035 adds work.graph_nodes.dispatch_id; migration runner applies it before runtime start"], + "failure_retry_restart": ["revisioned graph history survives process restart; dispatch queue claims remain recoverable by agent-worker"], + "assumptions": ["dispatch intents supplied with schedule request are already authorized by work-owned planning flow"], + "acceptance_scenarios": ["two independent nodes fan out under MaxParallel", "dependent join remains blocked until both predecessors succeed", "join completion yields graph COMPLETED", "cycle and missing dispatch bindings fail closed"], + "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "docker compose config"], + "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT", "REQUIRES_MIGRATION_DECISION"] +} diff --git a/modules/work/application/graph_runtime.go b/modules/work/application/graph_runtime.go new file mode 100644 index 0000000..77827b4 --- /dev/null +++ b/modules/work/application/graph_runtime.go @@ -0,0 +1,130 @@ +package application + +import ( + "context" + "errors" + "time" + + "git.cin.su/m/Assistant/contracts/work" + "git.cin.su/m/Assistant/modules/work/domain" + "git.cin.su/m/Assistant/modules/work/ports" +) + +var ErrGraphDispatchBinding = errors.New("graph node dispatch binding is missing") + +type GraphScheduleRequest struct { + Graph work.WorkGraph `json:"graph"` + Intents []work.AgentDispatchIntent `json:"intents"` + Now int64 `json:"now_unix_millis"` +} + +type GraphCompleteRequest struct { + GraphID string `json:"graph_id"` + Revision int64 `json:"revision"` + NodeID work.GraphNodeID `json:"node_id"` + Succeeded bool `json:"succeeded"` +} + +func AdmitAndSchedule(ctx context.Context, store ports.GraphStore, request GraphScheduleRequest) (work.WorkGraph, []work.GraphNodeID, error) { + if request.Now < 1 { + request.Now = time.Now().UnixMilli() + } + admitted, err := domain.AdmitGraph(request.Graph) + if err != nil { + return work.WorkGraph{}, nil, err + } + if err := store.SaveWorkGraph(ctx, admitted); err != nil { + return work.WorkGraph{}, nil, err + } + ready, err := domain.ScheduleReadyNodes(admitted, 0) + if err != nil { + return work.WorkGraph{}, nil, err + } + intentByDispatch := make(map[string]work.AgentDispatchIntent, len(request.Intents)) + for _, intent := range request.Intents { + intentByDispatch[intent.DispatchID] = intent + } + next := admitted + next.Revision++ + next.State = work.GraphRunning + selected := make(map[work.GraphNodeID]bool, len(ready)) + for _, nodeID := range ready { + selected[nodeID] = true + } + for i := range next.Nodes { + if !selected[next.Nodes[i].ID] { + continue + } + if next.Nodes[i].DispatchID == "" { + return work.WorkGraph{}, nil, ErrGraphDispatchBinding + } + intent, ok := intentByDispatch[next.Nodes[i].DispatchID] + if !ok { + return work.WorkGraph{}, nil, ErrGraphDispatchBinding + } + if err := store.SaveDispatchIntent(ctx, intent); err != nil { + return work.WorkGraph{}, nil, err + } + if err := store.EnqueueDispatch(ctx, work.AgentDispatchQueueItem{DispatchID: intent.DispatchID, State: work.AgentDispatchQueued, AvailableAtUnixMillis: request.Now, UpdatedAtUnixMillis: request.Now}); err != nil { + return work.WorkGraph{}, nil, err + } + next.Nodes[i].State = work.GraphNodeLeased + } + if err := store.SaveWorkGraph(ctx, next); err != nil { + return work.WorkGraph{}, nil, err + } + return next, ready, nil +} + +func CompleteGraphNode(ctx context.Context, store ports.GraphStore, graphID string, revision int64, nodeID work.GraphNodeID, success bool) (work.WorkGraph, error) { + graph, err := store.LoadWorkGraph(ctx, graphID, revision) + if err != nil { + return work.WorkGraph{}, err + } + found := false + for i := range graph.Nodes { + if graph.Nodes[i].ID == nodeID { + found = true + if success { + graph.Nodes[i].State = work.GraphNodeSuccess + } else { + graph.Nodes[i].State = work.GraphNodeFailed + } + } + } + if !found { + return work.WorkGraph{}, domain.ErrInvalidGraph + } + if !success { + graph.State = work.GraphFailed + } else { + ready, err := domain.ReadyGraphNodes(graph) + if err != nil { + return work.WorkGraph{}, err + } + for i := range graph.Nodes { + for _, candidate := range ready { + if graph.Nodes[i].ID == candidate && graph.Nodes[i].State == work.GraphNodeBlocked { + graph.Nodes[i].State = work.GraphNodeReady + } + } + } + allSuccess := true + for _, node := range graph.Nodes { + if node.State != work.GraphNodeSuccess { + allSuccess = false + break + } + } + if allSuccess { + graph.State = work.GraphCompleted + } else { + graph.State = work.GraphPartial + } + } + graph.Revision++ + if err := store.SaveWorkGraph(ctx, graph); err != nil { + return work.WorkGraph{}, err + } + return graph, nil +} diff --git a/modules/work/ports/graph.go b/modules/work/ports/graph.go new file mode 100644 index 0000000..1bb4612 --- /dev/null +++ b/modules/work/ports/graph.go @@ -0,0 +1,14 @@ +package ports + +import ( + "context" + + "git.cin.su/m/Assistant/contracts/work" +) + +type GraphStore interface { + SaveWorkGraph(context.Context, work.WorkGraph) error + LoadWorkGraph(context.Context, string, int64) (work.WorkGraph, error) + SaveDispatchIntent(context.Context, work.AgentDispatchIntent) error + EnqueueDispatch(context.Context, work.AgentDispatchQueueItem) error +} diff --git a/tests/agent_execution/graph_runtime_test.go b/tests/agent_execution/graph_runtime_test.go new file mode 100644 index 0000000..c7c99ea --- /dev/null +++ b/tests/agent_execution/graph_runtime_test.go @@ -0,0 +1,57 @@ +package agent_execution_test + +import ( + "context" + "strconv" + "testing" + + "git.cin.su/m/Assistant/contracts/work" + "git.cin.su/m/Assistant/modules/work/application" +) + +type graphStore struct { + graphs map[string]work.WorkGraph + queued []string + intents []string +} + +func (s *graphStore) SaveWorkGraph(_ context.Context, graph work.WorkGraph) error { + if s.graphs == nil { + s.graphs = map[string]work.WorkGraph{} + } + s.graphs[graph.ID+":"+itoa(graph.Revision)] = graph + return nil +} +func (s *graphStore) LoadWorkGraph(_ context.Context, id string, revision int64) (work.WorkGraph, error) { + return s.graphs[id+":"+itoa(revision)], nil +} +func (s *graphStore) SaveDispatchIntent(_ context.Context, intent work.AgentDispatchIntent) error { + s.intents = append(s.intents, intent.DispatchID) + return nil +} +func (s *graphStore) EnqueueDispatch(_ context.Context, item work.AgentDispatchQueueItem) error { + s.queued = append(s.queued, item.DispatchID) + return nil +} + +func itoa(value int64) string { + return strconv.FormatInt(value, 10) +} + +func TestGraphRuntimeSchedulesFanOutAndCompletesFanIn(t *testing.T) { + store := &graphStore{} + graph, ready, err := application.AdmitAndSchedule(context.Background(), store, application.GraphScheduleRequest{Graph: work.WorkGraph{ID: "g", Revision: 1, Budget: work.GraphBudget{MaxParallel: 2, MaxAttempts: 1}, Nodes: []work.GraphNode{{ID: "a", DispatchID: "d-a", IdempotencyKey: "a", State: work.GraphNodeReady}, {ID: "b", DispatchID: "d-b", IdempotencyKey: "b", State: work.GraphNodeReady}, {ID: "join", DispatchID: "d-join", IdempotencyKey: "join", State: work.GraphNodeBlocked}}, Edges: []work.GraphEdge{{From: "a", To: "join"}, {From: "b", To: "join"}}}, Intents: []work.AgentDispatchIntent{{DispatchID: "d-a"}, {DispatchID: "d-b"}}, Now: 1}) + if err != nil || len(ready) != 2 || len(store.queued) != 2 { + t.Fatalf("graph=%+v ready=%v queued=%v err=%v", graph, ready, store.queued, err) + } + if _, err := application.CompleteGraphNode(context.Background(), store, "g", 2, "a", true); err != nil { + t.Fatal(err) + } + if _, err := application.CompleteGraphNode(context.Background(), store, "g", 3, "b", true); err != nil { + t.Fatal(err) + } + completed, err := application.CompleteGraphNode(context.Background(), store, "g", 4, "join", true) + if err != nil || completed.State != work.GraphCompleted { + t.Fatalf("completed=%+v err=%v", completed, err) + } +} -- 2.54.0