diff --git a/internal/projectcontext/workflow_step.go b/internal/projectcontext/workflow_step.go index 048d808..186adbf 100644 --- a/internal/projectcontext/workflow_step.go +++ b/internal/projectcontext/workflow_step.go @@ -3,6 +3,9 @@ package projectcontext import "context" type workflowStepAttemptKey struct{} +type resolveScopeKey struct{} + +type resolveScope struct{ goal, targetID string } // WithWorkflowStepAttempt is used by the trusted squad executor adapter. The // value comes from the engine's persisted step attempt, never from model text. @@ -14,3 +17,15 @@ func WorkflowStepAttempt(ctx context.Context) string { value, _ := ctx.Value(workflowStepAttemptKey{}).(string) return value } + +// WithResolveScope carries the exact selector used for this model request to +// Backend's trusted receipt writer. It is saved separately from the public +// final-input manifest and survives a resumed workflow judge. +func WithResolveScope(ctx context.Context, goal, targetID string) context.Context { + return context.WithValue(ctx, resolveScopeKey{}, resolveScope{goal: goal, targetID: targetID}) +} + +func ResolveScope(ctx context.Context) (goal, targetID string, ok bool) { + scope, ok := ctx.Value(resolveScopeKey{}).(resolveScope) + return scope.goal, scope.targetID, ok +} diff --git a/internal/runmgr/project_context.go b/internal/runmgr/project_context.go index 072e671..763eb84 100644 --- a/internal/runmgr/project_context.go +++ b/internal/runmgr/project_context.go @@ -11,6 +11,7 @@ import ( agentllm "agent-native/runtime/llm" "agent-system-backend/internal/projectcontext" + "agent-system-backend/internal/store" "agent-system-backend/pkg/extension" core "agent-system-core" ) @@ -182,6 +183,10 @@ func (c *runCtx) SaveContextManifest(ctx context.Context, m extension.ProjectCon if run.Owner != c.run.Owner || run.ProjectID != c.run.ProjectID || run.ConversationID != c.run.ConversationID || m.ProjectID != run.ProjectID { return errors.New("context manifest run scope mismatch") } + resolveGoal, resolveTarget, scopeBound := projectcontext.ResolveScope(ctx) + if scopeBound && resolveTarget != "" { + return errors.New("run context does not support a model-selected Resolve target") + } // A response/unknown receipt records an already attempted transport. A // revocation after send must not erase that audit fact. if m.Delivery != "responded" && m.Delivery != "unknown" { @@ -191,6 +196,10 @@ func (c *runCtx) SaveContextManifest(ctx context.Context, m extension.ProjectCon } if c.mgr.projectContextResolver != nil && m.Delivery != "responded" && m.Delivery != "unknown" { resolvedReq := c.contextRequest() + if scopeBound { + resolvedReq = extension.ProjectContextRequest{Actor: string(run.Owner), ProjectID: run.ProjectID, + ConversationID: run.ConversationID, RunID: run.ID, Goal: resolveGoal, TargetID: resolveTarget, BudgetChars: m.BudgetChars} + } if resolvedReq.RunID != run.ID || resolvedReq.ProjectID != run.ProjectID { return errors.New("context manifest missing resolved run scope") } @@ -247,6 +256,10 @@ func (c *runCtx) SaveContextManifest(ctx context.Context, m extension.ProjectCon } } } + if scopeBound { + return c.mgr.st.AppendContextManifestWithScope(string(run.Owner), string(run.ID), m, + store.ContextManifestScope{Goal: resolveGoal, TargetID: resolveTarget, Bound: true}) + } return c.mgr.st.AppendContextManifest(string(run.Owner), string(run.ID), m) } @@ -266,17 +279,12 @@ func (c *runCtx) BindProjectStepOutput(ctx context.Context, stepID, stepAttemptI if kind != "prompt" || !c.ProjectContextEnabled() { return errors.New("project prompt step provenance unavailable") } - bundle, active, err := c.ProjectContext(ctx, "", 12000) - if err != nil || !active { - return errors.New("project step current scope unavailable") - } - bound, err := c.mgr.st.BindWorkflowStepSource(string(c.run.Owner), string(c.run.ID), stepID, stepAttemptID, kind, startedAt, output) + _, err := c.mgr.st.BindWorkflowStepSource(string(c.run.Owner), string(c.run.ID), stepID, stepAttemptID, kind, startedAt, output) if err != nil { return err } - if bound.ProjectID != string(bundle.ProjectID) || bound.AuthorizationEpoch != bundle.AuthorizationEpoch || - bound.PublicationEpoch != bundle.PublicationEpoch || bound.Watermark != bundle.Watermark { - return errors.New("project step source changed before binding") + if err := c.VerifyProjectStepOutput(ctx, stepID, stepAttemptID, kind, output); err != nil { + return errors.Join(err, c.mgr.st.UnbindWorkflowStepSource(string(c.run.Owner), string(c.run.ID), stepID)) } return nil } @@ -295,39 +303,53 @@ func (c *runCtx) VerifyProjectStepOutput(ctx context.Context, stepID, stepAttemp if bound.ProjectID != string(c.run.ProjectID) || bound.StepAttemptID != stepAttemptID || bound.OutputSHA256 != sha256.Sum256(output) { return errors.New("project workflow output bytes or project scope changed") } - receipt, err := c.mgr.st.RespondedContextManifest(string(c.run.Owner), string(c.run.ID), bound.AttemptID) - if err != nil || receipt.ProjectID != c.run.ProjectID || receipt.WorkflowStepAttemptID != stepAttemptID || receipt.AuthorizationEpoch != bound.AuthorizationEpoch || - receipt.PublicationEpoch != bound.PublicationEpoch || receipt.Watermark != bound.Watermark || receipt.Version != bound.Version { + receipts, err := c.mgr.st.RespondedWorkflowStepManifests(string(c.run.Owner), string(c.run.ID), stepAttemptID) + if err != nil { return errors.New("project workflow output lost its trusted model source") } - current, active, err := c.ProjectContext(ctx, "", 12000) - if err != nil || !active || current.ProjectID != receipt.ProjectID || - current.AuthorizationEpoch != receipt.AuthorizationEpoch || current.PublicationEpoch != receipt.PublicationEpoch || current.Watermark != receipt.Watermark { - return errors.New("project workflow output authorization or publication changed") - } - if err := projectcontext.ValidateCurrent(receipt, current); err != nil { - return fmt.Errorf("project workflow output source changed: %w", err) - } - for _, used := range receipt.ReadEvidence { - if c.mgr.projectEvidenceReader == nil { - return errors.New("project workflow evidence reader unavailable") + matchedBound := false + for _, source := range receipts { + receipt, selector := source.Manifest, source.Scope + if !selector.Bound || selector.TargetID != "" || receipt.ProjectID != c.run.ProjectID || + receipt.WorkflowStepAttemptID != stepAttemptID || receipt.AuthorizationEpoch != bound.AuthorizationEpoch || + receipt.PublicationEpoch != bound.PublicationEpoch || receipt.Watermark != bound.Watermark || receipt.Version != bound.Version { + return errors.New("project workflow output lost its trusted model source") } - matched := false - for _, ref := range current.Evidence { - if ref.ID != used.ReferenceID || ref.SourceID != used.SourceID || ref.Revision != used.Revision || ref.Hash != used.SourceHash { - continue + if receipt.AttemptID == bound.AttemptID { + matchedBound = true + } + current, active, err := c.ProjectContext(ctx, selector.Goal, receipt.BudgetChars) + if err != nil || !active || current.ProjectID != receipt.ProjectID || + current.AuthorizationEpoch != receipt.AuthorizationEpoch || current.PublicationEpoch != receipt.PublicationEpoch || current.Watermark != receipt.Watermark { + return errors.New("project workflow output authorization or publication changed") + } + if err := projectcontext.ValidateCurrent(receipt, current); err != nil { + return fmt.Errorf("project workflow output source changed: %w", err) + } + for _, used := range receipt.ReadEvidence { + if c.mgr.projectEvidenceReader == nil { + return errors.New("project workflow evidence reader unavailable") } - scope := c.contextRequest() - scope.BudgetChars = 8000 - now, err := c.mgr.projectEvidenceReader(ctx, extension.ProjectEvidenceRequest{Scope: scope, Reference: ref}) - if err == nil && now.Reference == ref && fmt.Sprintf("%x", sha256.Sum256([]byte(now.Text))) == used.InputSHA256 { - matched = true + matched := false + for _, ref := range current.Evidence { + if ref.ID != used.ReferenceID || ref.SourceID != used.SourceID || ref.Revision != used.Revision || ref.Hash != used.SourceHash { + continue + } + scope := extension.ProjectContextRequest{Actor: string(c.run.Owner), ProjectID: c.run.ProjectID, + ConversationID: c.run.ConversationID, RunID: c.run.ID, Goal: selector.Goal, TargetID: selector.TargetID, BudgetChars: 8000} + now, err := c.mgr.projectEvidenceReader(ctx, extension.ProjectEvidenceRequest{Scope: scope, Reference: ref}) + if err == nil && now.Reference == ref && fmt.Sprintf("%x", sha256.Sum256([]byte(now.Text))) == used.InputSHA256 { + matched = true + } + break + } + if !matched { + return errors.New("project workflow evidence permission or content changed") } - break - } - if !matched { - return errors.New("project workflow evidence permission or content changed") } } + if !matchedBound { + return errors.New("project workflow output binding lost its responded model attempt") + } return nil } diff --git a/internal/runmgr/project_context_test.go b/internal/runmgr/project_context_test.go index c18aee1..290802f 100644 --- a/internal/runmgr/project_context_test.go +++ b/internal/runmgr/project_context_test.go @@ -2,12 +2,19 @@ package runmgr import ( "context" + "fmt" + "os" + "strings" "sync" "testing" "time" agentllm "agent-native/runtime/llm" + "agent-squad/squad" "agent-system-backend/internal/projectcontext" + "agent-system-backend/internal/runtime" + "agent-system-backend/internal/wfcompile" + "agent-system-backend/internal/workflow" "agent-system-backend/pkg/extension" core "agent-system-core" ) @@ -95,11 +102,12 @@ func TestProjectJudgeChecksPersistedSourceEpochAcrossGuards(t *testing.T) { } manifest.InputSHA256, manifest.AttemptID, manifest.Delivery = "input-hash", "step-attempt", "attempted" manifest.WorkflowStepAttemptID = "engine-run/draft/1" - if err := rc.SaveContextManifest(context.Background(), manifest); err != nil { + sourceCtx := projectcontext.WithResolveScope(context.Background(), "write output", "") + if err := rc.SaveContextManifest(sourceCtx, manifest); err != nil { t.Fatal(err) } manifest.Delivery = "responded" - if err := rc.SaveContextManifest(context.Background(), manifest); err != nil { + if err := rc.SaveContextManifest(sourceCtx, manifest); err != nil { t.Fatal(err) } if err := rc.BindProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", started, []byte("OLD_RESTRICTED_BODY")); err != nil { @@ -158,6 +166,148 @@ func TestProjectJudgeChecksPersistedSourceEpochAcrossGuards(t *testing.T) { } } +func TestProjectJudgeRetainsGoalSensitiveSourceScope(t *testing.T) { + st := openRecoveryStore(t) + m := newRecoveryManager(t, st) + conv := recoveryConversation(t, st) + run := recoveryRun(t, st, conv, core.RunPending) + m.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return true, nil }) + available := true + var goals []string + m.SetProjectContextResolver(func(_ context.Context, req extension.ProjectContextRequest) (extension.ProjectContextBundle, error) { + goals = append(goals, req.Goal) + item := extension.ProjectContextItem{SourceID: "source-b", Revision: "r1", Hash: "hb", Text: "PUBLIC_B", Required: true} + if strings.Contains(req.Goal, "draft report") && available { + item = extension.ProjectContextItem{SourceID: "source-a", Revision: "r1", Hash: "ha", Text: "AUTHORIZED_A", Required: true} + } + return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID, AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Watermark: "w1", Items: []extension.ProjectContextItem{item}}, nil + }) + rc := &runCtx{mgr: m, run: run, ctx: context.Background()} + const goal = "draft report from approved source" + bundle, _, err := rc.ProjectContext(context.Background(), goal, 900) + if err != nil { + t.Fatal(err) + } + _, receipt, err := projectcontext.Assemble(bundle, 900) + if err != nil { + t.Fatal(err) + } + receipt.InputSHA256, receipt.AttemptID, receipt.WorkflowStepAttemptID = "input-hash", "model-attempt", "engine-run/draft/1" + receipt.Delivery = "attempted" + sourceCtx := projectcontext.WithResolveScope(context.Background(), goal, "") + if err := rc.SaveContextManifest(sourceCtx, receipt); err != nil { + t.Fatal(err) + } + receipt.Delivery = "responded" + if err := rc.SaveContextManifest(sourceCtx, receipt); err != nil { + t.Fatal(err) + } + // The final model turn uses Source B; output may still carry Source A from + // the first turn. Revalidate both receipts at the critic boundary. + bundleB, _, err := rc.ProjectContext(context.Background(), "revise using B", 900) + if err != nil { + t.Fatal(err) + } + _, second, err := projectcontext.Assemble(bundleB, 900) + if err != nil { + t.Fatal(err) + } + second.InputSHA256, second.AttemptID, second.WorkflowStepAttemptID = "second-input-hash", "model-attempt-2", "engine-run/draft/1" + secondCtx := projectcontext.WithResolveScope(context.Background(), "revise using B", "") + second.Delivery = "attempted" + if err := rc.SaveContextManifest(secondCtx, second); err != nil { + t.Fatal(err) + } + second.Delivery = "responded" + if err := rc.SaveContextManifest(secondCtx, second); err != nil { + t.Fatal(err) + } + started := time.Now().Add(-time.Second) + if err := rc.BindProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", started, []byte("report A")); err != nil { + t.Fatalf("same-scope binding rejected: %v", err) + } + if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("report A")); err != nil { + t.Fatalf("goal-sensitive provider rejected authorized critic: %v; goals=%q", err, goals) + } + available = false + if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("report A")); err == nil { + t.Fatal("withdrawn source passed goal-sensitive revalidation") + } +} + +type goalSensitiveStepBackend struct { + stream agentllm.StreamFunc + sent string +} + +func (*goalSensitiveStepBackend) Name() string { return "goal-sensitive-test" } +func (b *goalSensitiveStepBackend) Execute(ctx context.Context, req squad.ExecRequest) (*squad.ExecResult, error) { + for ev := range b.stream(ctx, "test-model", []agentllm.Message{{Role: agentllm.RoleUser, Content: req.Prompt}}, nil, agentllm.StreamOptions{}) { + if ev.Kind == agentllm.KindError { + return &squad.ExecResult{Success: false, ErrorReason: ev.Err.Error()}, nil + } + } + b.sent = req.Prompt + if err := os.WriteFile(req.WorkDir+"/output.md", []byte("AUTHORIZED_A report\n"), 0o600); err != nil { + return nil, err + } + return &squad.ExecResult{Success: true, Patches: map[string][]byte{"output.md": []byte("AUTHORIZED_A report\n")}}, nil +} + +func TestCompiledWorkflowJudgeAcceptsGoalSensitiveProviderAndRejectsWithdrawal(t *testing.T) { + st := openRecoveryStore(t) + mgr := newRecoveryManager(t, st) + conv := recoveryConversation(t, st) + run := recoveryRun(t, st, conv, core.RunPending) + mgr.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return true, nil }) + available := true + var resolved []string + mgr.SetProjectContextResolver(func(_ context.Context, req extension.ProjectContextRequest) (extension.ProjectContextBundle, error) { + resolved = append(resolved, req.Goal) + item := extension.ProjectContextItem{SourceID: "source-b", Revision: "r1", Hash: "hb", Text: "PUBLIC_B", Required: true} + if strings.Contains(req.Goal, "draft report") && !strings.Contains(req.Goal, "你是验收员") && available { + item = extension.ProjectContextItem{SourceID: "source-a", Revision: "r1", Hash: "ha", Text: "AUTHORIZED_A", Required: true} + } + return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID, AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Watermark: "w1", Items: []extension.ProjectContextItem{item}}, nil + }) + rc := &runCtx{mgr: mgr, run: run, ctx: context.Background()} + stepBase := func(_ context.Context, _ string, messages []agentllm.Message, _ []agentllm.ToolDecl, _ agentllm.StreamOptions) <-chan agentllm.Event { + ch := make(chan agentllm.Event, 1) + if !strings.Contains(messages[len(messages)-1].Content, "AUTHORIZED_A") { + ch <- agentllm.Event{Kind: agentllm.KindError, Err: fmt.Errorf("draft model did not receive source A")} + } else { + ch <- agentllm.Event{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: "drafted"}} + } + close(ch) + return ch + } + judgeCalls := 0 + judgeBase := func(_ context.Context, _ string, messages []agentllm.Message, _ []agentllm.ToolDecl, _ agentllm.StreamOptions) <-chan agentllm.Event { + judgeCalls++ + ch := make(chan agentllm.Event, 1) + ch <- agentllm.Event{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: `{"passed":true,"reason":"ok","nextAction":""}`}} + close(ch) + return ch + } + backend := &goalSensitiveStepBackend{stream: runtime.GuardProjectStream(rc, "", stepBase)} + backends := squad.NewBackendResolver() + backends.RegisterInstance(backend.Name(), backend) + def := core.WorkflowDef{ID: "goal-sensitive", Title: "Report", Steps: []core.WorkflowStep{{ID: "draft", Title: "Draft", Kind: "prompt", Agent: "writer", Prompt: "draft report using the approved source", Critic: &core.StepCritic{Criteria: "contains report"}}}} + _, err := wfcompile.RunCompiled(rc, wfcompile.Deps{DataDir: t.TempDir(), Backends: backends, DefaultBackend: backend.Name(), JudgeStream: runtime.GuardProjectStream(rc, "", judgeBase)}, + workflow.Launch{Def: def}, workflow.Vars{Input: "draft report"}, wfcompile.RunOptions{}) + if err != nil || judgeCalls != 1 || backend.sent == "" { + t.Fatalf("authorized goal-sensitive workflow failed: err=%v judgeCalls=%d sent=%q resolved=%q", err, judgeCalls, backend.sent, resolved) + } + bound, err := st.WorkflowStepSource(string(run.Owner), string(run.ID), "draft") + if err != nil { + t.Fatal(err) + } + available = false + if err := rc.VerifyProjectStepOutput(context.Background(), "draft", bound.StepAttemptID, "prompt", []byte("AUTHORIZED_A report\n")); err == nil { + t.Fatal("withdrawn source A passed resumed goal-sensitive judge") + } +} + func TestContextManifestPersistsByOwnerAndRun(t *testing.T) { st := openRecoveryStore(t) m := newRecoveryManager(t, st) diff --git a/internal/runtime/project_stream.go b/internal/runtime/project_stream.go index aae73ae..31fd97c 100644 --- a/internal/runtime/project_stream.go +++ b/internal/runtime/project_stream.go @@ -146,7 +146,9 @@ func GuardProjectStream(rc any, goal string, stream agentllm.StreamFunc) agentll if sink, ok := rc.(interface { SaveContextManifest(context.Context, extension.ProjectContextManifest) error }); ok { - save = sink.SaveContextManifest + save = func(saveCtx context.Context, receipt extension.ProjectContextManifest) error { + return sink.SaveContextManifest(projectcontext.WithResolveScope(saveCtx, requestGoal, ""), receipt) + } if err := save(ctx, manifest); err != nil { return fail(err) } diff --git a/internal/store/context_manifest.go b/internal/store/context_manifest.go index 5179f6b..779f7b2 100644 --- a/internal/store/context_manifest.go +++ b/internal/store/context_manifest.go @@ -11,10 +11,29 @@ import ( const contextManifestAction = "model.context_manifest" +// ContextManifestScope records the exact trusted Resolve selector separately +// from the public final-input manifest. Bound distinguishes an intentionally +// empty goal from a legacy receipt with no recoverable selector. +type ContextManifestScope struct { + Goal, TargetID string + Bound bool +} + // AppendContextManifest writes to the trusted receipt table, never to the // plugin-writable generic audit log. A terminal state must follow a matching // attempted state for the same owner, target, project and input hash. func (s *Store) AppendContextManifest(owner, targetID string, m extension.ProjectContextManifest) error { + return s.appendContextManifest(owner, targetID, m, ContextManifestScope{}) +} + +func (s *Store) AppendContextManifestWithScope(owner, targetID string, m extension.ProjectContextManifest, scope ContextManifestScope) error { + if !scope.Bound { + return errors.New("trusted context manifest resolve scope is not bound") + } + return s.appendContextManifest(owner, targetID, m, scope) +} + +func (s *Store) appendContextManifest(owner, targetID string, m extension.ProjectContextManifest, scope ContextManifestScope) error { if owner == "" || targetID == "" || m.ProjectID == "" || m.InputSHA256 == "" || m.AttemptID == "" { return errors.New("context manifest requires owner, target, project, attempt and input hash") } @@ -31,15 +50,19 @@ func (s *Store) AppendContextManifest(owner, targetID string, m extension.Projec } defer tx.Rollback() if m.Delivery != "attempted" { - var found int - err := tx.QueryRow(s.q(`SELECT 1 FROM context_manifests WHERE attempt_id=? AND delivery='attempted' AND owner=? AND target_id=? AND project_id=? AND input_sha256=?`), - m.AttemptID, owner, targetID, m.ProjectID, m.InputSHA256).Scan(&found) + var attemptedGoal, attemptedTarget string + var attemptedBound int + err := tx.QueryRow(s.q(`SELECT resolve_goal,resolve_target_id,scope_bound FROM context_manifests WHERE attempt_id=? AND delivery='attempted' AND owner=? AND target_id=? AND project_id=? AND input_sha256=?`), + m.AttemptID, owner, targetID, m.ProjectID, m.InputSHA256).Scan(&attemptedGoal, &attemptedTarget, &attemptedBound) if errors.Is(err, sql.ErrNoRows) { return errors.New("context manifest terminal receipt lacks matching attempt") } if err != nil { return err } + if attemptedGoal != scope.Goal || attemptedTarget != scope.TargetID || (attemptedBound != 0) != scope.Bound { + return errors.New("context manifest terminal resolve scope changed") + } var terminal int err = tx.QueryRow(s.q(`SELECT 1 FROM context_manifests WHERE attempt_id=? AND delivery IN ('responded','unknown') LIMIT 1`), m.AttemptID).Scan(&terminal) if err == nil { @@ -49,8 +72,12 @@ func (s *Store) AppendContextManifest(owner, targetID string, m extension.Projec return err } } - _, err = tx.Exec(s.q(`INSERT INTO context_manifests(id,attempt_id,owner,target_id,project_id,input_sha256,delivery,manifest_json,created_at) VALUES(?,?,?,?,?,?,?,?,?)`), - NewID("ctxm"), m.AttemptID, owner, targetID, m.ProjectID, m.InputSHA256, m.Delivery, string(raw), nowISO()) + bound := 0 + if scope.Bound { + bound = 1 + } + _, err = tx.Exec(s.q(`INSERT INTO context_manifests(id,attempt_id,owner,target_id,project_id,input_sha256,delivery,manifest_json,created_at,resolve_goal,resolve_target_id,scope_bound) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`), + NewID("ctxm"), m.AttemptID, owner, targetID, m.ProjectID, m.InputSHA256, m.Delivery, string(raw), nowISO(), scope.Goal, scope.TargetID, bound) if err != nil { return fmt.Errorf("insert context manifest: %w", err) } diff --git a/internal/store/context_manifest_test.go b/internal/store/context_manifest_test.go index 7d9018d..c4411b8 100644 --- a/internal/store/context_manifest_test.go +++ b/internal/store/context_manifest_test.go @@ -36,3 +36,26 @@ func TestContextManifestTerminalRequiresMatchingAttempt(t *testing.T) { t.Fatal("one attempt acquired two terminal states") } } + +func TestContextManifestTerminalCannotChangeTrustedResolveScope(t *testing.T) { + st, err := OpenSQLite(filepath.Join(t.TempDir(), "scope.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + m := extension.ProjectContextManifest{ProjectID: "p1", AttemptID: "model-1", WorkflowStepAttemptID: "engine/draft/1", InputSHA256: "input-hash", Delivery: "attempted"} + scope := ContextManifestScope{Goal: "draft from A", Bound: true} + if err := st.AppendContextManifestWithScope("alice", "run-1", m, scope); err != nil { + t.Fatal(err) + } + m.Delivery = "responded" + if err := st.AppendContextManifestWithScope("alice", "run-1", m, ContextManifestScope{Goal: "draft from B", Bound: true}); err == nil { + t.Fatal("terminal receipt changed its trusted Resolve goal") + } + if err := st.AppendContextManifest("alice", "run-1", m); err == nil { + t.Fatal("terminal receipt dropped its trusted Resolve scope") + } + if err := st.AppendContextManifestWithScope("alice", "run-1", m, scope); err != nil { + t.Fatal(err) + } +} diff --git a/internal/store/migrate.go b/internal/store/migrate.go index f5ab653..bf6fab9 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -869,6 +869,16 @@ func init() { }, down: map[string][]string{"sqlite": {`DROP TABLE IF EXISTS workflow_step_sources`}, "postgres": {`DROP TABLE IF EXISTS workflow_step_sources`}}, }, + migration{version: 34, name: "context_manifest_resolve_scope", + up: map[string][]string{ + "sqlite": {`ALTER TABLE context_manifests ADD COLUMN resolve_goal TEXT NOT NULL DEFAULT ''`, `ALTER TABLE context_manifests ADD COLUMN resolve_target_id TEXT NOT NULL DEFAULT ''`, `ALTER TABLE context_manifests ADD COLUMN scope_bound INTEGER NOT NULL DEFAULT 0`}, + "postgres": {`ALTER TABLE context_manifests ADD COLUMN resolve_goal TEXT NOT NULL DEFAULT ''`, `ALTER TABLE context_manifests ADD COLUMN resolve_target_id TEXT NOT NULL DEFAULT ''`, `ALTER TABLE context_manifests ADD COLUMN scope_bound INTEGER NOT NULL DEFAULT 0`}, + }, + down: map[string][]string{ + "sqlite": {`ALTER TABLE context_manifests DROP COLUMN scope_bound`, `ALTER TABLE context_manifests DROP COLUMN resolve_target_id`, `ALTER TABLE context_manifests DROP COLUMN resolve_goal`}, + "postgres": {`ALTER TABLE context_manifests DROP COLUMN scope_bound`, `ALTER TABLE context_manifests DROP COLUMN resolve_target_id`, `ALTER TABLE context_manifests DROP COLUMN resolve_goal`}, + }, + }, ) } diff --git a/internal/store/migrate_compat.go b/internal/store/migrate_compat.go index efb7cb2..7b17ce6 100644 --- a/internal/store/migrate_compat.go +++ b/internal/store/migrate_compat.go @@ -38,6 +38,7 @@ var migrationBreaksOldReaders = map[int]bool{ 31: false, // work_task_bindings：新增绑定表，旧后台任务读面不受影响 32: false, // trusted_context_manifests：独立可信回执表，旧读者不受影响 33: false, // workflow_step_sources：新增逐步骤来源绑定表，旧读者不受影响 + 34: false, // context_manifest_resolve_scope：可信回执增加原Resolve范围，旧读者不受影响 } // SchemaTarget 是本二进制会迁移到的 schema 版本。 diff --git a/internal/store/workflow_step_source.go b/internal/store/workflow_step_source.go index 3ff9efd..e5bbfad 100644 --- a/internal/store/workflow_step_source.go +++ b/internal/store/workflow_step_source.go @@ -40,7 +40,7 @@ func (s *Store) BindWorkflowStepSource(owner, runID, stepID, stepAttemptID, kind if err != nil || run.Owner != owner || run.ProjectID == "" { return WorkflowStepSource{}, ErrNotFound } - rows, err := s.db.Query(s.q(`SELECT created_at,manifest_json FROM context_manifests WHERE owner=? AND target_id=? AND delivery='responded'`), owner, runID) + rows, err := s.db.Query(s.q(`SELECT created_at,manifest_json,resolve_target_id,scope_bound FROM context_manifests WHERE owner=? AND target_id=? AND delivery='responded'`), owner, runID) if err != nil { return WorkflowStepSource{}, err } @@ -48,8 +48,9 @@ func (s *Store) BindWorkflowStepSource(owner, runID, stepID, stepAttemptID, kind var latest extension.ProjectContextManifest var latestAt time.Time for rows.Next() { - var atRaw, raw string - if err := rows.Scan(&atRaw, &raw); err != nil { + var atRaw, raw, resolveTarget string + var scopeBound int + if err := rows.Scan(&atRaw, &raw, &resolveTarget, &scopeBound); err != nil { return WorkflowStepSource{}, err } at, err := time.Parse(time.RFC3339Nano, atRaw) @@ -66,10 +67,13 @@ func (s *Store) BindWorkflowStepSource(owner, runID, stepID, stepAttemptID, kind if m.WorkflowStepAttemptID != stepAttemptID { continue } + if scopeBound != 1 || resolveTarget != "" { + return WorkflowStepSource{}, errors.New("workflow step model receipt lacks trusted original Resolve scope") + } if m.ProjectID != run.ProjectID || m.AuthorizationEpoch == "" || m.PublicationEpoch == "" || m.AttemptID == "" { return WorkflowStepSource{}, errors.New("workflow step model receipt lacks trusted project epochs") } - if !latestAt.IsZero() && (m.ProjectID != latest.ProjectID || m.AuthorizationEpoch != latest.AuthorizationEpoch || m.PublicationEpoch != latest.PublicationEpoch || m.Watermark != latest.Watermark) { + if !latestAt.IsZero() && (m.ProjectID != latest.ProjectID || m.AuthorizationEpoch != latest.AuthorizationEpoch || m.PublicationEpoch != latest.PublicationEpoch || m.Watermark != latest.Watermark || m.Version != latest.Version) { return WorkflowStepSource{}, errors.New("workflow step model receipts span project authorization revisions") } if latestAt.IsZero() || at.After(latestAt) { @@ -92,6 +96,48 @@ func (s *Store) BindWorkflowStepSource(owner, runID, stepID, stepAttemptID, kind return b, nil } +type ScopedContextManifest struct { + Manifest extension.ProjectContextManifest + Scope ContextManifestScope +} + +// RespondedWorkflowStepManifests returns every model response from this exact +// engine step attempt. A final output can depend on earlier model turns, so a +// critic must revalidate all of them, not just the last response. +func (s *Store) RespondedWorkflowStepManifests(owner, runID, stepAttemptID string) ([]ScopedContextManifest, error) { + if owner == "" || runID == "" || stepAttemptID == "" { + return nil, errors.New("workflow receipt scope is incomplete") + } + rows, err := s.db.Query(s.q(`SELECT manifest_json,resolve_goal,resolve_target_id,scope_bound FROM context_manifests WHERE owner=? AND target_id=? AND delivery='responded' ORDER BY created_at,id`), owner, runID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []ScopedContextManifest + for rows.Next() { + var raw, goal, target string + var bound int + if err := rows.Scan(&raw, &goal, &target, &bound); err != nil { + return nil, err + } + var manifest extension.ProjectContextManifest + if err := json.Unmarshal([]byte(raw), &manifest); err != nil { + return nil, err + } + if manifest.WorkflowStepAttemptID != stepAttemptID { + continue + } + out = append(out, ScopedContextManifest{Manifest: manifest, Scope: ContextManifestScope{Goal: goal, TargetID: target, Bound: bound == 1}}) + } + if err := rows.Err(); err != nil { + return nil, err + } + if len(out) == 0 { + return nil, ErrNotFound + } + return out, nil +} + func (s *Store) WorkflowStepSource(owner, runID, stepID string) (WorkflowStepSource, error) { var b WorkflowStepSource var hash string diff --git a/internal/store/workflow_step_source_test.go b/internal/store/workflow_step_source_test.go index 2c68602..f58d9f7 100644 --- a/internal/store/workflow_step_source_test.go +++ b/internal/store/workflow_step_source_test.go @@ -30,7 +30,8 @@ func TestWorkflowStepSourceRequiresRespondedManifestAndExactRunStep(t *testing.T } started := time.Now().Add(-time.Second) m := extension.ProjectContextManifest{ProjectID: p.ID, AttemptID: "attempt-1", WorkflowStepAttemptID: "engine-run/draft/1", Delivery: "attempted", InputSHA256: "input", AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Watermark: "w1"} - if err := st.AppendContextManifest("alice", string(r.ID), m); err != nil { + scope := ContextManifestScope{Goal: "draft report", Bound: true} + if err := st.AppendContextManifestWithScope("alice", string(r.ID), m, scope); err != nil { t.Fatal(err) } if _, err := st.BindWorkflowStepSource("alice", string(r.ID), "draft", "engine-run/draft/1", "prompt", started, []byte("old body")); err == nil { @@ -39,18 +40,18 @@ func TestWorkflowStepSourceRequiresRespondedManifestAndExactRunStep(t *testing.T unrelated := m unrelated.AttemptID = "other-model-attempt" unrelated.WorkflowStepAttemptID = "engine-run/other/1" - if err := st.AppendContextManifest("alice", string(r.ID), unrelated); err != nil { + if err := st.AppendContextManifestWithScope("alice", string(r.ID), unrelated, scope); err != nil { t.Fatal(err) } unrelated.Delivery = "responded" - if err := st.AppendContextManifest("alice", string(r.ID), unrelated); err != nil { + if err := st.AppendContextManifestWithScope("alice", string(r.ID), unrelated, scope); err != nil { t.Fatal(err) } if _, err := st.BindWorkflowStepSource("alice", string(r.ID), "draft", "engine-run/draft/1", "prompt", started, []byte("old body")); err == nil { t.Fatal("another step's responded attempt bound draft output") } m.Delivery = "responded" - if err := st.AppendContextManifest("alice", string(r.ID), m); err != nil { + if err := st.AppendContextManifestWithScope("alice", string(r.ID), m, scope); err != nil { t.Fatal(err) } bound, err := st.BindWorkflowStepSource("alice", string(r.ID), "draft", "engine-run/draft/1", "prompt", started, []byte("old body")) @@ -66,6 +67,18 @@ func TestWorkflowStepSourceRequiresRespondedManifestAndExactRunStep(t *testing.T if _, err := st.BindWorkflowStepSource("alice", string(r.ID), "tool", "engine-run/tool/1", "tool", started, []byte("tool body")); err == nil { t.Fatal("tool output bound as model-sourced material") } + legacy := m + legacy.AttemptID, legacy.WorkflowStepAttemptID, legacy.Delivery = "legacy-model", "engine-run/legacy/1", "attempted" + if err := st.AppendContextManifest("alice", string(r.ID), legacy); err != nil { + t.Fatal(err) + } + legacy.Delivery = "responded" + if err := st.AppendContextManifest("alice", string(r.ID), legacy); err != nil { + t.Fatal(err) + } + if _, err := st.BindWorkflowStepSource("alice", string(r.ID), "legacy", "engine-run/legacy/1", "prompt", started, []byte("legacy body")); err == nil { + t.Fatal("model receipt without bound original Resolve scope became a workflow source") + } if err := st.UnbindWorkflowStepSource("alice", string(r.ID), "draft"); err != nil { t.Fatal(err) }