Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
def0e4aff0 | ||
|
|
9a6ab6a9ca | ||
|
|
8f20326c07 | ||
|
|
a82592a130 |
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
"git.cin.su/m/Assistant/contracts/work"
|
||||
@@ -106,24 +107,28 @@ func (r *WorkRepository) SaveDispatchIntent(ctx context.Context, intent work.Age
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
_, err = tx.Exec(ctx, `INSERT INTO work.agent_dispatch_intents(dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) ON CONFLICT(idempotency_key) DO NOTHING`, intent.DispatchID, intent.IdempotencyKey, intent.DelegationID, intent.RunID, intent.ImplementationID, intent.ImplementationRevision, intent.SourceRevision, intent.WorktreeID, intent.VerificationPlanRef, intent.CreatedAtUnixMillis)
|
||||
_, err = tx.Exec(ctx, `INSERT INTO work.agent_dispatch_intents(dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis,specialist_purpose,task_evaluation_profile_id,task_evaluation_binding_id,task_evaluation_case_ids) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT(idempotency_key) DO NOTHING`, intent.DispatchID, intent.IdempotencyKey, intent.DelegationID, intent.RunID, intent.ImplementationID, intent.ImplementationRevision, intent.SourceRevision, intent.WorktreeID, intent.VerificationPlanRef, intent.CreatedAtUnixMillis, intent.SpecialistPurpose, intent.TaskEvaluationProfileID, intent.TaskEvaluationBindingID, intent.TaskEvaluationCaseIDs)
|
||||
if err != nil {
|
||||
return fmt.Errorf("save dispatch intent: %w", err)
|
||||
}
|
||||
var existing work.AgentDispatchIntent
|
||||
err = tx.QueryRow(ctx, `SELECT dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis FROM work.agent_dispatch_intents WHERE idempotency_key=$1`, intent.IdempotencyKey).Scan(&existing.DispatchID, &existing.IdempotencyKey, &existing.DelegationID, &existing.RunID, &existing.ImplementationID, &existing.ImplementationRevision, &existing.SourceRevision, &existing.WorktreeID, &existing.VerificationPlanRef, &existing.CreatedAtUnixMillis)
|
||||
err = tx.QueryRow(ctx, `SELECT dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis,specialist_purpose,task_evaluation_profile_id,task_evaluation_binding_id,task_evaluation_case_ids FROM work.agent_dispatch_intents WHERE idempotency_key=$1`, intent.IdempotencyKey).Scan(&existing.DispatchID, &existing.IdempotencyKey, &existing.DelegationID, &existing.RunID, &existing.ImplementationID, &existing.ImplementationRevision, &existing.SourceRevision, &existing.WorktreeID, &existing.VerificationPlanRef, &existing.CreatedAtUnixMillis, &existing.SpecialistPurpose, &existing.TaskEvaluationProfileID, &existing.TaskEvaluationBindingID, &existing.TaskEvaluationCaseIDs)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read dispatch intent: %w", err)
|
||||
}
|
||||
if existing != intent {
|
||||
if !sameDispatchIntent(existing, intent) {
|
||||
return ErrDispatchIntentConflict
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
|
||||
func sameDispatchIntent(left, right work.AgentDispatchIntent) bool {
|
||||
return left.DispatchID == right.DispatchID && left.IdempotencyKey == right.IdempotencyKey && left.DelegationID == right.DelegationID && left.RunID == right.RunID && left.ImplementationID == right.ImplementationID && left.ImplementationRevision == right.ImplementationRevision && left.SourceRevision == right.SourceRevision && left.WorktreeID == right.WorktreeID && left.VerificationPlanRef == right.VerificationPlanRef && left.CreatedAtUnixMillis == right.CreatedAtUnixMillis && left.SpecialistPurpose == right.SpecialistPurpose && left.TaskEvaluationProfileID == right.TaskEvaluationProfileID && left.TaskEvaluationBindingID == right.TaskEvaluationBindingID && slices.Equal(left.TaskEvaluationCaseIDs, right.TaskEvaluationCaseIDs)
|
||||
}
|
||||
|
||||
func (r *WorkRepository) LoadDispatchIntent(ctx context.Context, dispatchID string) (work.AgentDispatchIntent, error) {
|
||||
var intent work.AgentDispatchIntent
|
||||
err := r.pool.QueryRow(ctx, `SELECT dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis FROM work.agent_dispatch_intents WHERE dispatch_id=$1`, dispatchID).Scan(&intent.DispatchID, &intent.IdempotencyKey, &intent.DelegationID, &intent.RunID, &intent.ImplementationID, &intent.ImplementationRevision, &intent.SourceRevision, &intent.WorktreeID, &intent.VerificationPlanRef, &intent.CreatedAtUnixMillis)
|
||||
err := r.pool.QueryRow(ctx, `SELECT dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis,specialist_purpose,task_evaluation_profile_id,task_evaluation_binding_id,task_evaluation_case_ids FROM work.agent_dispatch_intents WHERE dispatch_id=$1`, dispatchID).Scan(&intent.DispatchID, &intent.IdempotencyKey, &intent.DelegationID, &intent.RunID, &intent.ImplementationID, &intent.ImplementationRevision, &intent.SourceRevision, &intent.WorktreeID, &intent.VerificationPlanRef, &intent.CreatedAtUnixMillis, &intent.SpecialistPurpose, &intent.TaskEvaluationProfileID, &intent.TaskEvaluationBindingID, &intent.TaskEvaluationCaseIDs)
|
||||
if err != nil {
|
||||
return work.AgentDispatchIntent{}, fmt.Errorf("load dispatch intent: %w", err)
|
||||
}
|
||||
@@ -191,11 +196,11 @@ func (r *WorkRepository) SaveWorkGraph(ctx context.Context, graph work.WorkGraph
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
if _, err := tx.Exec(ctx, `INSERT INTO work.graphs(graph_id,revision,state,max_parallel,max_attempts) VALUES($1,$2,$3,$4,$5) ON CONFLICT(graph_id,revision) DO NOTHING`, graph.ID, graph.Revision, graph.State, graph.Budget.MaxParallel, graph.Budget.MaxAttempts); err != nil {
|
||||
if _, err := tx.Exec(ctx, `INSERT INTO work.graphs(graph_id,revision,state,max_parallel,max_attempts,parent_graph_id,parent_graph_revision,trigger_evidence_ref) VALUES($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT(graph_id,revision) DO NOTHING`, graph.ID, graph.Revision, graph.State, graph.Budget.MaxParallel, graph.Budget.MaxAttempts, graph.ParentGraphID, graph.ParentGraphRevision, graph.TriggerEvidenceRef); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, node := range graph.Nodes {
|
||||
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 {
|
||||
if _, err := tx.Exec(ctx, `INSERT INTO work.graph_nodes(graph_id,revision,node_id,dispatch_id,idempotency_key,state,attempts,purpose,task_profile_id,binding_id,case_ids) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) ON CONFLICT(graph_id,revision,node_id) DO NOTHING`, graph.ID, graph.Revision, node.ID, node.DispatchID, node.IdempotencyKey, node.State, node.Attempts, node.Purpose, node.TaskProfileID, node.BindingID, node.CaseIDs); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
@@ -209,17 +214,17 @@ func (r *WorkRepository) SaveWorkGraph(ctx context.Context, graph work.WorkGraph
|
||||
|
||||
func (r *WorkRepository) LoadWorkGraph(ctx context.Context, graphID string, revision int64) (work.WorkGraph, error) {
|
||||
var graph work.WorkGraph
|
||||
err := r.pool.QueryRow(ctx, `SELECT graph_id,revision,state,max_parallel,max_attempts FROM work.graphs WHERE graph_id=$1 AND revision=$2`, graphID, revision).Scan(&graph.ID, &graph.Revision, &graph.State, &graph.Budget.MaxParallel, &graph.Budget.MaxAttempts)
|
||||
err := r.pool.QueryRow(ctx, `SELECT graph_id,revision,state,max_parallel,max_attempts,parent_graph_id,parent_graph_revision,trigger_evidence_ref FROM work.graphs WHERE graph_id=$1 AND revision=$2`, graphID, revision).Scan(&graph.ID, &graph.Revision, &graph.State, &graph.Budget.MaxParallel, &graph.Budget.MaxAttempts, &graph.ParentGraphID, &graph.ParentGraphRevision, &graph.TriggerEvidenceRef)
|
||||
if err != nil {
|
||||
return work.WorkGraph{}, fmt.Errorf("load graph: %w", err)
|
||||
}
|
||||
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)
|
||||
rows, err := r.pool.Query(ctx, `SELECT node_id,dispatch_id,idempotency_key,state,attempts,purpose,task_profile_id,binding_id,case_ids 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.DispatchID, &node.IdempotencyKey, &node.State, &node.Attempts); err != nil {
|
||||
if err := rows.Scan(&node.ID, &node.DispatchID, &node.IdempotencyKey, &node.State, &node.Attempts, &node.Purpose, &node.TaskProfileID, &node.BindingID, &node.CaseIDs); err != nil {
|
||||
rows.Close()
|
||||
return work.WorkGraph{}, err
|
||||
}
|
||||
|
||||
@@ -42,6 +42,19 @@ func main() {
|
||||
}
|
||||
writeJSON(w, map[string]any{"graph": graph, "scheduled_nodes": ready})
|
||||
})
|
||||
mux.HandleFunc("POST /v1/work-graphs/task-evaluations/schedule", func(w http.ResponseWriter, r *http.Request) {
|
||||
var request application.TaskEvaluationScheduleRequest
|
||||
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&request); err != nil {
|
||||
http.Error(w, "invalid task evaluation graph request", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
graph, ready, err := application.AdmitAndScheduleTaskEvaluation(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 {
|
||||
|
||||
+40
-11
@@ -1,11 +1,17 @@
|
||||
package work
|
||||
|
||||
import "git.cin.su/m/Assistant/contracts/capability"
|
||||
|
||||
type GraphNodeID string
|
||||
|
||||
type GraphState string
|
||||
type GraphNodeState string
|
||||
type GraphNodePurpose string
|
||||
|
||||
const (
|
||||
GraphNodePurposeTaskEvaluation GraphNodePurpose = "TASK_EVALUATION_SPECIALIST"
|
||||
GraphNodePurposeGeneric GraphNodePurpose = "GENERIC"
|
||||
|
||||
GraphDraft GraphState = "DRAFT"
|
||||
GraphAdmitted GraphState = "ADMITTED"
|
||||
GraphRunning GraphState = "RUNNING"
|
||||
@@ -31,11 +37,15 @@ type GraphBudget struct {
|
||||
}
|
||||
|
||||
type GraphNode struct {
|
||||
ID GraphNodeID `json:"id"`
|
||||
DispatchID string `json:"dispatch_id"`
|
||||
IdempotencyKey string `json:"idempotency_key"`
|
||||
State GraphNodeState `json:"state"`
|
||||
Attempts int `json:"attempts"`
|
||||
ID GraphNodeID `json:"id"`
|
||||
DispatchID string `json:"dispatch_id"`
|
||||
IdempotencyKey string `json:"idempotency_key"`
|
||||
State GraphNodeState `json:"state"`
|
||||
Attempts int `json:"attempts"`
|
||||
Purpose GraphNodePurpose `json:"purpose,omitempty"`
|
||||
TaskProfileID capability.TaskEvaluationProfileID `json:"task_profile_id,omitempty"`
|
||||
BindingID capability.CapabilityBindingID `json:"binding_id,omitempty"`
|
||||
CaseIDs []string `json:"case_ids,omitempty"`
|
||||
}
|
||||
|
||||
type GraphEdge struct {
|
||||
@@ -44,12 +54,31 @@ type GraphEdge struct {
|
||||
}
|
||||
|
||||
type WorkGraph struct {
|
||||
ID string `json:"id"`
|
||||
Revision int64 `json:"revision"`
|
||||
State GraphState `json:"state"`
|
||||
Nodes []GraphNode `json:"nodes"`
|
||||
Edges []GraphEdge `json:"edges"`
|
||||
Budget GraphBudget `json:"budget"`
|
||||
ID string `json:"id"`
|
||||
Revision int64 `json:"revision"`
|
||||
State GraphState `json:"state"`
|
||||
Nodes []GraphNode `json:"nodes"`
|
||||
Edges []GraphEdge `json:"edges"`
|
||||
Budget GraphBudget `json:"budget"`
|
||||
ParentGraphID string `json:"parent_graph_id,omitempty"`
|
||||
ParentGraphRevision int64 `json:"parent_graph_revision,omitempty"`
|
||||
TriggerEvidenceRef string `json:"trigger_evidence_ref,omitempty"`
|
||||
}
|
||||
|
||||
// TaskEvaluationGraphRequest describes a durable fan-out. The graph contains
|
||||
// one specialist node per binding; each node receives the profile's bounded
|
||||
// cases and reports evidence back to Capability for fan-in aggregation.
|
||||
type TaskEvaluationGraphRequest struct {
|
||||
GraphID string `json:"graph_id"`
|
||||
Profile capability.TaskEvaluationProfile `json:"profile"`
|
||||
BindingIDs []capability.CapabilityBindingID `json:"binding_ids"`
|
||||
DispatchIDs []string `json:"dispatch_ids"`
|
||||
IdempotencyPrefix string `json:"idempotency_prefix"`
|
||||
Budget GraphBudget `json:"budget"`
|
||||
ParentGraphID string `json:"parent_graph_id,omitempty"`
|
||||
ParentGraphRevision int64 `json:"parent_graph_revision,omitempty"`
|
||||
ParentGraphState GraphState `json:"parent_graph_state,omitempty"`
|
||||
TriggerEvidenceRef string `json:"trigger_evidence_ref,omitempty"`
|
||||
}
|
||||
|
||||
type ResultCandidate struct {
|
||||
|
||||
+19
-11
@@ -1,6 +1,10 @@
|
||||
package work
|
||||
|
||||
import "time"
|
||||
import (
|
||||
"time"
|
||||
|
||||
"git.cin.su/m/Assistant/contracts/capability"
|
||||
)
|
||||
|
||||
type AgentLeaseID string
|
||||
type AgentFencingToken uint64
|
||||
@@ -50,14 +54,18 @@ type IndependentVerification struct {
|
||||
}
|
||||
|
||||
type AgentDispatchIntent struct {
|
||||
DispatchID string `json:"dispatch_id"`
|
||||
IdempotencyKey string `json:"idempotency_key"`
|
||||
DelegationID DelegationID `json:"delegation_id"`
|
||||
RunID AgentRunID `json:"run_id"`
|
||||
ImplementationID string `json:"implementation_id"`
|
||||
ImplementationRevision int64 `json:"implementation_revision"`
|
||||
SourceRevision string `json:"source_revision"`
|
||||
WorktreeID string `json:"worktree_id"`
|
||||
VerificationPlanRef string `json:"verification_plan_ref"`
|
||||
CreatedAtUnixMillis int64 `json:"created_at_unix_millis"`
|
||||
DispatchID string `json:"dispatch_id"`
|
||||
IdempotencyKey string `json:"idempotency_key"`
|
||||
DelegationID DelegationID `json:"delegation_id"`
|
||||
RunID AgentRunID `json:"run_id"`
|
||||
ImplementationID string `json:"implementation_id"`
|
||||
ImplementationRevision int64 `json:"implementation_revision"`
|
||||
SourceRevision string `json:"source_revision"`
|
||||
WorktreeID string `json:"worktree_id"`
|
||||
VerificationPlanRef string `json:"verification_plan_ref"`
|
||||
CreatedAtUnixMillis int64 `json:"created_at_unix_millis"`
|
||||
SpecialistPurpose string `json:"specialist_purpose,omitempty"`
|
||||
TaskEvaluationProfileID capability.TaskEvaluationProfileID `json:"task_evaluation_profile_id,omitempty"`
|
||||
TaskEvaluationBindingID capability.CapabilityBindingID `json:"task_evaluation_binding_id,omitempty"`
|
||||
TaskEvaluationCaseIDs []string `json:"task_evaluation_case_ids,omitempty"`
|
||||
}
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
BEGIN;
|
||||
|
||||
SET ROLE work_role;
|
||||
|
||||
ALTER TABLE work.agent_dispatch_intents
|
||||
ADD COLUMN IF NOT EXISTS specialist_purpose TEXT NOT NULL DEFAULT '',
|
||||
ADD COLUMN IF NOT EXISTS task_evaluation_profile_id TEXT NOT NULL DEFAULT '',
|
||||
ADD COLUMN IF NOT EXISTS task_evaluation_binding_id TEXT NOT NULL DEFAULT '',
|
||||
ADD COLUMN IF NOT EXISTS task_evaluation_case_ids TEXT[] NOT NULL DEFAULT '{}';
|
||||
|
||||
ALTER TABLE work.graphs
|
||||
ADD COLUMN IF NOT EXISTS parent_graph_id TEXT NOT NULL DEFAULT '',
|
||||
ADD COLUMN IF NOT EXISTS parent_graph_revision BIGINT NOT NULL DEFAULT 0,
|
||||
ADD COLUMN IF NOT EXISTS trigger_evidence_ref TEXT NOT NULL DEFAULT '';
|
||||
|
||||
ALTER TABLE work.graph_nodes
|
||||
ADD COLUMN IF NOT EXISTS purpose TEXT NOT NULL DEFAULT 'GENERIC',
|
||||
ADD COLUMN IF NOT EXISTS task_profile_id TEXT NOT NULL DEFAULT '',
|
||||
ADD COLUMN IF NOT EXISTS binding_id TEXT NOT NULL DEFAULT '',
|
||||
ADD COLUMN IF NOT EXISTS case_ids TEXT[] NOT NULL DEFAULT '{}';
|
||||
|
||||
RESET ROLE;
|
||||
COMMIT;
|
||||
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"task_file": "dev/tasks/DEV-000208-task-aware-model-evaluation.json",
|
||||
"task_id": "DEV-000208",
|
||||
"task_file": "dev/tasks/DEV-000210-task-evaluator-worker.json",
|
||||
"task_id": "DEV-000210",
|
||||
"task_revision": 1
|
||||
}
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"task_id": "DEV-000209",
|
||||
"task_revision": 1,
|
||||
"architecture_revision": "ADR-0018",
|
||||
"base_repo_revision": "a82592a1300c06ed7a568c058faca6560e085ab3",
|
||||
"risk_tier": "E",
|
||||
"intent": "Create durable Work Graph plans for task-aware model evaluation: run the general profile first, fan out independent model specialists in parallel, and permit bounded additional profiles linked to unresolved evidence without blocking the chat path.",
|
||||
"requirement_ids": ["general-first model evaluation", "parallel specialist evaluation", "bounded additional task tests", "fan-out/fan-in", "no synchronous chat blocking", "provider and hardware neutral"],
|
||||
"allowed_change_scope": {"modules": ["work"], "paths": ["contracts/work/**", "modules/work/**", "cmd/work-graph-runtime/**", "tests/model_qualification/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000209-task-evaluator-work-graph.json"]},
|
||||
"forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "cmd/devctl/**", "modules/identity/**", "database/**"]},
|
||||
"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": ["capability.TaskEvaluationProfile", "capability.TaskEvaluationProfileStage", "capability.TaskEvaluationProfileID", "existing Work Graph schedule contract"], "may_change": ["typed Work Graph node purpose and task-evaluation planning contract"]},
|
||||
"semantic_owner": "work",
|
||||
"canonical_state": ["work-owned durable graph and node lifecycle; capability remains owner of model evaluation evidence"],
|
||||
"state_transitions": ["GENERAL profile -> admitted evaluation graph", "eligible model bindings -> parallel specialist nodes", "all specialist nodes -> fan-in completion state", "ADDITIONAL profile -> a separate bounded graph linked by parent/trigger evidence"],
|
||||
"side_effects_and_privileges": ["none; graph planning only, no provider calls and no secret access"],
|
||||
"persistence_and_migrations": ["uses existing work graph persistence; no new schema"],
|
||||
"failure_retry_restart": ["each specialist remains independently retryable through existing node attempts/leases; partial or unknown graph state is preserved", "invalid stage/parent linkage is rejected before scheduling"],
|
||||
"assumptions": ["a later worker adapter executes each dispatched specialist and posts evidence to capability", "chat/request handlers enqueue the graph and return without waiting for all nodes"],
|
||||
"acceptance_scenarios": ["additional profile cannot be scheduled as a replacement for general profile", "one node is created per binding and profile with independent dispatch identity", "parallel budget is retained for fan-out", "graph metadata is sufficient for a worker to identify profile, binding and specialist role", "provider kind, endpoint and hardware are absent from planning inputs"],
|
||||
"required_tests": ["go test", "go vet", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "CI", "production public health check"],
|
||||
"terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_MISSING_CAPABILITY"]
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"task_id": "DEV-000210",
|
||||
"task_revision": 1,
|
||||
"architecture_revision": "ADR-0018",
|
||||
"base_repo_revision": "9a6ab6a9caf68e65f302c95ab0859b259122dacf",
|
||||
"risk_tier": "E",
|
||||
"intent": "Carry task-evaluation specialist metadata through durable dispatch, persistence and recovery so workers can execute bounded model cases and return evidence without losing profile lineage.",
|
||||
"requirement_ids": ["worker execution context", "durable task metadata", "recovery-safe dispatch", "bounded case set", "no synchronous chat blocking"],
|
||||
"allowed_change_scope": {"modules": ["work"], "paths": ["contracts/work/**", "modules/work/**", "adapters/postgres/work_repository.go", "cmd/agent-worker/**", "database/migrations/040_task_evaluation_dispatch.sql", "tests/model_qualification/**", "tests/agent_execution/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000210-task-evaluator-worker.json"]},
|
||||
"forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "cmd/devctl/**", "modules/identity/**", "modules/capability/**"]},
|
||||
"allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 1},
|
||||
"contracts": {"consume": ["Work Graph task-evaluation node metadata", "AgentExecutionRequest/Result", "existing lease and recovery contract"], "may_change": ["typed dispatch task-evaluation metadata"]},
|
||||
"semantic_owner": "work",
|
||||
"canonical_state": ["work-owned dispatch metadata and graph lineage; capability remains owner of evaluation evidence"],
|
||||
"state_transitions": ["graph specialist node -> enriched dispatch intent", "claimed dispatch -> executor receives profile/binding/cases", "stale claim -> recovery preserves task metadata for retry"],
|
||||
"side_effects_and_privileges": ["work role migration only; no provider calls and no secret access added"],
|
||||
"persistence_and_migrations": ["040 adds typed task metadata columns to existing work dispatch/graph tables"],
|
||||
"failure_retry_restart": ["metadata is persisted before enqueue; idempotency conflict rejects divergent task context", "lease timeout/recovery does not erase profile or case references"],
|
||||
"assumptions": ["an upcoming cognitive specialist executor uses the typed metadata to invoke the model gateway and post Capability evidence", "external bridge absence remains typed WAITING/local-fallback"],
|
||||
"acceptance_scenarios": ["task graph dispatch carries profile, binding and all cases", "persist/load round trip preserves metadata", "divergent idempotency retry is rejected", "legacy non-task dispatch remains valid", "recovery path does not claim success"],
|
||||
"required_tests": ["go test", "go vet", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "migration/schema checks", "CI", "production public health check"],
|
||||
"terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_MISSING_CAPABILITY"]
|
||||
}
|
||||
@@ -18,6 +18,12 @@ type GraphScheduleRequest struct {
|
||||
Now int64 `json:"now_unix_millis"`
|
||||
}
|
||||
|
||||
type TaskEvaluationScheduleRequest struct {
|
||||
Plan work.TaskEvaluationGraphRequest `json:"plan"`
|
||||
Intents []work.AgentDispatchIntent `json:"intents"`
|
||||
Now int64 `json:"now_unix_millis"`
|
||||
}
|
||||
|
||||
type GraphCompleteRequest struct {
|
||||
GraphID string `json:"graph_id"`
|
||||
Revision int64 `json:"revision"`
|
||||
@@ -105,6 +111,32 @@ func AdmitAndSchedule(ctx context.Context, store ports.GraphStore, request Graph
|
||||
return next, ready, nil
|
||||
}
|
||||
|
||||
// AdmitAndScheduleTaskEvaluation creates the durable general-first or
|
||||
// bounded-expansion fan-out before handing it to the existing graph scheduler.
|
||||
// The request path only persists/enqueues work; specialist execution happens
|
||||
// independently through the dispatch workers.
|
||||
func AdmitAndScheduleTaskEvaluation(ctx context.Context, store ports.GraphStore, request TaskEvaluationScheduleRequest) (work.WorkGraph, []work.GraphNodeID, error) {
|
||||
graph, err := domain.BuildTaskEvaluationGraph(request.Plan)
|
||||
if err != nil {
|
||||
return work.WorkGraph{}, nil, err
|
||||
}
|
||||
intents := make([]work.AgentDispatchIntent, len(request.Intents))
|
||||
copy(intents, request.Intents)
|
||||
for i := range intents {
|
||||
for _, node := range graph.Nodes {
|
||||
if node.DispatchID != intents[i].DispatchID {
|
||||
continue
|
||||
}
|
||||
intents[i].SpecialistPurpose = string(node.Purpose)
|
||||
intents[i].TaskEvaluationProfileID = node.TaskProfileID
|
||||
intents[i].TaskEvaluationBindingID = node.BindingID
|
||||
intents[i].TaskEvaluationCaseIDs = append([]string(nil), node.CaseIDs...)
|
||||
break
|
||||
}
|
||||
}
|
||||
return AdmitAndSchedule(ctx, store, GraphScheduleRequest{Graph: graph, Intents: intents, Now: request.Now})
|
||||
}
|
||||
|
||||
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 {
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
package application
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"git.cin.su/m/Assistant/contracts/capability"
|
||||
"git.cin.su/m/Assistant/contracts/work"
|
||||
)
|
||||
|
||||
type taskGraphStore struct {
|
||||
graph work.WorkGraph
|
||||
intents []work.AgentDispatchIntent
|
||||
}
|
||||
|
||||
func (s *taskGraphStore) SaveWorkGraph(_ context.Context, graph work.WorkGraph) error {
|
||||
s.graph = graph
|
||||
return nil
|
||||
}
|
||||
func (s *taskGraphStore) LoadWorkGraph(context.Context, string, int64) (work.WorkGraph, error) {
|
||||
return s.graph, nil
|
||||
}
|
||||
func (s *taskGraphStore) SaveDispatchIntent(_ context.Context, intent work.AgentDispatchIntent) error {
|
||||
s.intents = append(s.intents, intent)
|
||||
return nil
|
||||
}
|
||||
func (s *taskGraphStore) EnqueueDispatch(context.Context, work.AgentDispatchQueueItem) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestAdmitAndScheduleTaskEvaluationCarriesSpecialistContext(t *testing.T) {
|
||||
store := &taskGraphStore{}
|
||||
request := TaskEvaluationScheduleRequest{
|
||||
Plan: work.TaskEvaluationGraphRequest{
|
||||
GraphID: "graph-task", Profile: capability.TaskEvaluationProfile{ID: "profile-general", Revision: 1, CapabilityID: "capability-chat", Stage: capability.TaskEvaluationGeneral, BenchmarkVersion: "general-v1", RequiredCaseIDs: []string{"case-a", "case-b"}, QualityWeightMillis: 600, ReliabilityWeightMillis: 300, LatencyWeightMillis: 100},
|
||||
BindingIDs: []capability.CapabilityBindingID{"binding-a"}, DispatchIDs: []string{"dispatch-a"}, IdempotencyPrefix: "eval", Budget: work.GraphBudget{MaxParallel: 1, MaxAttempts: 2},
|
||||
},
|
||||
Intents: []work.AgentDispatchIntent{{DispatchID: "dispatch-a", IdempotencyKey: "idem-a", DelegationID: "delegation-a", RunID: "run-a", ImplementationID: "impl", ImplementationRevision: 1, SourceRevision: "source", WorktreeID: "tree", VerificationPlanRef: "verify", CreatedAtUnixMillis: 1}},
|
||||
}
|
||||
if _, _, err := AdmitAndScheduleTaskEvaluation(context.Background(), store, request); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(store.intents) != 1 || store.intents[0].TaskEvaluationProfileID != "profile-general" || store.intents[0].TaskEvaluationBindingID != "binding-a" || len(store.intents[0].TaskEvaluationCaseIDs) != 2 || store.intents[0].SpecialistPurpose != string(work.GraphNodePurposeTaskEvaluation) {
|
||||
t.Fatalf("specialist context was not carried into dispatch: %+v", store.intents)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
package domain
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"git.cin.su/m/Assistant/contracts/capability"
|
||||
"git.cin.su/m/Assistant/contracts/work"
|
||||
)
|
||||
|
||||
var ErrInvalidTaskEvaluationGraph = errors.New("invalid task evaluation graph request")
|
||||
|
||||
// BuildTaskEvaluationGraph turns one bounded profile into durable specialist
|
||||
// nodes. It deliberately has no provider, endpoint or hardware inputs: those
|
||||
// are resolved later by the cognitive worker for each binding.
|
||||
func BuildTaskEvaluationGraph(request work.TaskEvaluationGraphRequest) (work.WorkGraph, error) {
|
||||
if request.GraphID == "" || request.IdempotencyPrefix == "" || len(request.BindingIDs) == 0 || len(request.BindingIDs) != len(request.DispatchIDs) || request.Budget.MaxParallel < 1 || request.Budget.MaxAttempts < 1 {
|
||||
return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph
|
||||
}
|
||||
if err := request.Profile.Validate(); err != nil {
|
||||
return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph
|
||||
}
|
||||
if request.Profile.Stage == capability.TaskEvaluationGeneral {
|
||||
if request.ParentGraphID != "" || request.ParentGraphRevision != 0 || request.ParentGraphState != "" || request.TriggerEvidenceRef != "" {
|
||||
return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph
|
||||
}
|
||||
} else {
|
||||
if request.ParentGraphID == "" || request.ParentGraphRevision < 1 || request.TriggerEvidenceRef == "" || (request.ParentGraphState != work.GraphCompleted && request.ParentGraphState != work.GraphPartial) {
|
||||
return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph
|
||||
}
|
||||
if request.Profile.ParentProfileID == "" {
|
||||
return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph
|
||||
}
|
||||
}
|
||||
seenBindings := make(map[capability.CapabilityBindingID]struct{}, len(request.BindingIDs))
|
||||
seenDispatches := make(map[string]struct{}, len(request.DispatchIDs))
|
||||
nodes := make([]work.GraphNode, 0, len(request.BindingIDs))
|
||||
for i, bindingID := range request.BindingIDs {
|
||||
if bindingID == "" || request.DispatchIDs[i] == "" {
|
||||
return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph
|
||||
}
|
||||
if _, ok := seenBindings[bindingID]; ok {
|
||||
return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph
|
||||
}
|
||||
if _, ok := seenDispatches[request.DispatchIDs[i]]; ok {
|
||||
return work.WorkGraph{}, ErrInvalidTaskEvaluationGraph
|
||||
}
|
||||
seenBindings[bindingID] = struct{}{}
|
||||
seenDispatches[request.DispatchIDs[i]] = struct{}{}
|
||||
nodes = append(nodes, work.GraphNode{
|
||||
ID: work.GraphNodeID(fmt.Sprintf("%s-node-%d", request.GraphID, i+1)),
|
||||
DispatchID: request.DispatchIDs[i],
|
||||
IdempotencyKey: fmt.Sprintf("%s:%s:%d", request.IdempotencyPrefix, bindingID, request.Profile.Revision),
|
||||
State: work.GraphNodeBlocked,
|
||||
Purpose: work.GraphNodePurposeTaskEvaluation,
|
||||
TaskProfileID: request.Profile.ID,
|
||||
BindingID: bindingID,
|
||||
CaseIDs: append([]string(nil), request.Profile.RequiredCaseIDs...),
|
||||
})
|
||||
}
|
||||
return work.WorkGraph{ID: request.GraphID, Revision: 1, State: work.GraphDraft, Nodes: nodes, Budget: request.Budget, ParentGraphID: request.ParentGraphID, ParentGraphRevision: request.ParentGraphRevision, TriggerEvidenceRef: request.TriggerEvidenceRef}, nil
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
package domain
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"git.cin.su/m/Assistant/contracts/capability"
|
||||
"git.cin.su/m/Assistant/contracts/work"
|
||||
)
|
||||
|
||||
func evaluationGraphRequest(stage capability.TaskEvaluationProfileStage) work.TaskEvaluationGraphRequest {
|
||||
profile := capability.TaskEvaluationProfile{ID: "profile-1", Revision: 1, CapabilityID: "capability-1", Stage: stage, BenchmarkVersion: "general-v1", RequiredCaseIDs: []string{"case-a", "case-b"}, QualityWeightMillis: 600, ReliabilityWeightMillis: 300, LatencyWeightMillis: 100}
|
||||
request := work.TaskEvaluationGraphRequest{GraphID: "graph-1", Profile: profile, BindingIDs: []capability.CapabilityBindingID{"binding-a", "binding-b"}, DispatchIDs: []string{"dispatch-a", "dispatch-b"}, IdempotencyPrefix: "task-eval", Budget: work.GraphBudget{MaxParallel: 2, MaxAttempts: 2}}
|
||||
if stage == capability.TaskEvaluationAdditional {
|
||||
request.Profile.ParentProfileID = "profile-general"
|
||||
request.Profile.TriggerEvidenceRef = "artifact:unresolved-1"
|
||||
request.Profile.Stage = capability.TaskEvaluationAdditional
|
||||
request.ParentGraphID, request.ParentGraphRevision, request.ParentGraphState, request.TriggerEvidenceRef = "graph-general", 2, work.GraphPartial, "artifact:unresolved-1"
|
||||
}
|
||||
return request
|
||||
}
|
||||
|
||||
func TestBuildTaskEvaluationGraphFansOutOneNodePerBinding(t *testing.T) {
|
||||
graph, err := BuildTaskEvaluationGraph(evaluationGraphRequest(capability.TaskEvaluationGeneral))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(graph.Nodes) != 2 || graph.Nodes[0].Purpose != work.GraphNodePurposeTaskEvaluation || graph.Nodes[0].TaskProfileID != "profile-1" || len(graph.Nodes[0].CaseIDs) != 2 {
|
||||
t.Fatalf("unexpected task graph nodes: %+v", graph.Nodes)
|
||||
}
|
||||
if graph.ParentGraphID != "" || graph.TriggerEvidenceRef != "" {
|
||||
t.Fatalf("general graph unexpectedly has expansion lineage: %+v", graph)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildTaskEvaluationGraphRequiresCompletedOrPartialGeneralParent(t *testing.T) {
|
||||
request := evaluationGraphRequest(capability.TaskEvaluationAdditional)
|
||||
request.ParentGraphState = work.GraphRunning
|
||||
if _, err := BuildTaskEvaluationGraph(request); err == nil {
|
||||
t.Fatal("expected additional graph to require a finished parent graph")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildTaskEvaluationGraphRejectsGeneralReplacement(t *testing.T) {
|
||||
request := evaluationGraphRequest(capability.TaskEvaluationGeneral)
|
||||
request.ParentGraphID = "old-graph"
|
||||
request.ParentGraphRevision = 2
|
||||
request.ParentGraphState = work.GraphPartial
|
||||
request.TriggerEvidenceRef = "artifact:unresolved"
|
||||
if _, err := BuildTaskEvaluationGraph(request); err == nil {
|
||||
t.Fatal("expected general graph replacement to be rejected")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
package agent_execution
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestTaskEvaluationDispatchMigrationPersistsWorkerContext(t *testing.T) {
|
||||
payload, err := os.ReadFile("../../database/migrations/040_task_evaluation_dispatch.sql")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sql := strings.ToLower(string(payload))
|
||||
for _, marker := range []string{"work.agent_dispatch_intents", "specialist_purpose", "task_evaluation_profile_id", "task_evaluation_binding_id", "task_evaluation_case_ids", "work.graphs", "parent_graph_id", "work.graph_nodes", "case_ids", "set role work_role"} {
|
||||
if !strings.Contains(sql, strings.ToLower(marker)) {
|
||||
t.Fatalf("migration missing %q", marker)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user