From 7a00b20cd4282c62aa4560085a804ea8c2a60349 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 14 Oct 2026 04:55:14 +0000 Subject: [PATCH] Bind workflow judge inputs to trusted step attempts Why: Compiled workflow critics could resend a prior prompt output after document duty or publication changed across separate model guards. Evidence: RED cross-guard revocation and retry tests; go test ./... -count=1 and targeted go test -race passed locally before commit. Impact: Bind materialized prompt bytes to the engine step attempt and responded model receipt; recheck current project sources and epochs before judging, failing closed for unproven or ambiguous output. Refs: RDM two-trees stage5 W6; stage6 X1 public extension request; W6 independent review P1. --- cmd/storage-migrate/main.go | 4 +- internal/projectcontext/workflow_step.go | 16 +++ internal/runmgr/project_context.go | 107 ++++++++++++++++ internal/runmgr/project_context_test.go | 88 ++++++++++++++ internal/runtime/project_stream.go | 8 ++ internal/runtime/workflow_compiled.go | 32 +++++ internal/runtime/workflow_compiled_test.go | 65 +++++++++- internal/store/migrate.go | 7 ++ internal/store/migrate_compat.go | 1 + internal/store/workflow_step_source.go | 128 ++++++++++++++++++++ internal/store/workflow_step_source_test.go | 75 ++++++++++++ internal/wfcompile/host.go | 32 ++++- internal/wfcompile/project_backend.go | 37 ++++++ internal/wfcompile/project_sources.go | 107 ++++++++++++++++ internal/wfcompile/run.go | 21 +++- internal/wfcompile/wfcompile_test.go | 116 ++++++++++++++++++ pkg/extension/project_context.go | 33 ++--- 17 files changed, 852 insertions(+), 25 deletions(-) create mode 100644 internal/projectcontext/workflow_step.go create mode 100644 internal/store/workflow_step_source.go create mode 100644 internal/store/workflow_step_source_test.go create mode 100644 internal/wfcompile/project_backend.go create mode 100644 internal/wfcompile/project_sources.go diff --git a/cmd/storage-migrate/main.go b/cmd/storage-migrate/main.go index 49206c6..92de9f2 100644 --- a/cmd/storage-migrate/main.go +++ b/cmd/storage-migrate/main.go @@ -287,7 +287,7 @@ var copyOrder = []string{ "activity_cards", "project_event_shares", "command_audit", "prompt_modules", "sessions", "api_keys", "plugin_grants", "plugin_schema", "plugin_kv", "background_tasks", "work_task_bindings", - "memory_entries", "audit_events", "context_manifests", "pending_confirms", "llm_usage_daily", + "memory_entries", "audit_events", "context_manifests", "workflow_step_sources", "pending_confirms", "llm_usage_daily", "secret_keys", "settings_snapshots", "regen_pending", "context_entries", "user_commands", "pending_notices", "recovery_holds", "surfaces", "surface_revisions", "surface_actions", "plugin_activation", @@ -784,7 +784,7 @@ func advanceSequences(db *sql.DB, table string) error { // checkOwners：业务行引用的 owner 必须存在于 users（单租户默认 ”/'local' // 除外）。未知 owner 阻塞迁移（U4/R7：不静默带过未知归属）。 func checkOwners(db *sql.DB, src *sql.DB) error { - ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards", "project_event_shares", "command_audit", "work_task_bindings", "context_manifests"} + ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards", "project_event_shares", "command_audit", "work_task_bindings", "context_manifests", "workflow_step_sources"} for _, t := range ownerTables { if !sqliteHasTable(src, t) { continue diff --git a/internal/projectcontext/workflow_step.go b/internal/projectcontext/workflow_step.go new file mode 100644 index 0000000..048d808 --- /dev/null +++ b/internal/projectcontext/workflow_step.go @@ -0,0 +1,16 @@ +package projectcontext + +import "context" + +type workflowStepAttemptKey struct{} + +// WithWorkflowStepAttempt is used by the trusted squad executor adapter. The +// value comes from the engine's persisted step attempt, never from model text. +func WithWorkflowStepAttempt(ctx context.Context, attemptID string) context.Context { + return context.WithValue(ctx, workflowStepAttemptKey{}, attemptID) +} + +func WorkflowStepAttempt(ctx context.Context) string { + value, _ := ctx.Value(workflowStepAttemptKey{}).(string) + return value +} diff --git a/internal/runmgr/project_context.go b/internal/runmgr/project_context.go index 5236b45..072e671 100644 --- a/internal/runmgr/project_context.go +++ b/internal/runmgr/project_context.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "strings" + "time" agentllm "agent-native/runtime/llm" @@ -28,6 +29,30 @@ func (c *runCtx) ProjectContextEnabled() bool { return c.run.ProjectID != "" && c.mgr.projectContextResolver != nil } +// ValidateProjectCrossGuard links separate model streams in one persisted run. +// A new guard cannot treat a workflow/user prompt as fresh when earlier model +// work under the same run used another authorization or publication revision. +func (c *runCtx) ValidateProjectCrossGuard(ctx context.Context, current extension.ProjectContextBundle) error { + if !c.ProjectContextEnabled() { + return nil + } + if err := ctx.Err(); err != nil { + return err + } + prior, err := c.mgr.st.ListContextManifests(string(c.run.Owner), string(c.run.ID)) + if err != nil { + return err + } + for _, m := range prior { + if m.ProjectID != current.ProjectID || m.AuthorizationEpoch == "" || m.PublicationEpoch == "" || + current.AuthorizationEpoch == "" || current.PublicationEpoch == "" || + m.AuthorizationEpoch != current.AuthorizationEpoch || m.PublicationEpoch != current.PublicationEpoch || m.Watermark != current.Watermark { + return errors.New("prior project model work used a different authorization or publication revision") + } + } + return nil +} + func (c *runCtx) contextRequest() extension.ProjectContextRequest { c.projectContextMu.Lock() defer c.projectContextMu.Unlock() @@ -224,3 +249,85 @@ func (c *runCtx) SaveContextManifest(ctx context.Context, m extension.ProjectCon } return c.mgr.st.AppendContextManifest(string(run.Owner), string(run.ID), m) } + +// InvalidateProjectStepOutput prevents a retry from inheriting the prior +// attempt's binding, even when the new attempt produces identical bytes. +func (c *runCtx) InvalidateProjectStepOutput(ctx context.Context, stepID string) error { + if err := ctx.Err(); err != nil { + return err + } + return c.mgr.st.UnbindWorkflowStepSource(string(c.run.Owner), string(c.run.ID), stepID) +} + +// BindProjectStepOutput stores exact bytes and their responded model attempt +// after a prompt step has materialized. The step start timestamp fences out +// receipts from earlier steps; the workflow sink rejects concurrent ambiguity. +func (c *runCtx) BindProjectStepOutput(ctx context.Context, stepID, stepAttemptID, kind string, startedAt time.Time, output []byte) error { + 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) + 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") + } + return nil +} + +// VerifyProjectStepOutput is the last gate before a workflow critic receives +// output.md. It checks the persisted run+step association, current file bytes, +// the trusted responded request, and live domain authorization epochs. +func (c *runCtx) VerifyProjectStepOutput(ctx context.Context, stepID, stepAttemptID, kind string, output []byte) error { + if kind != "prompt" || !c.ProjectContextEnabled() { + return errors.New("project workflow output has no verifiable prompt source") + } + bound, err := c.mgr.st.WorkflowStepSource(string(c.run.Owner), string(c.run.ID), stepID) + if err != nil { + return errors.New("project workflow output has no trusted step binding") + } + 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 { + 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") + } + 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 := 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 + } + break + } + if !matched { + return errors.New("project workflow evidence permission or content changed") + } + } + return nil +} diff --git a/internal/runmgr/project_context_test.go b/internal/runmgr/project_context_test.go index 3177358..c18aee1 100644 --- a/internal/runmgr/project_context_test.go +++ b/internal/runmgr/project_context_test.go @@ -4,6 +4,7 @@ import ( "context" "sync" "testing" + "time" agentllm "agent-native/runtime/llm" "agent-system-backend/internal/projectcontext" @@ -70,6 +71,93 @@ func TestProjectContextResumeReloadsCurrentVersionAndAuthorization(t *testing.T) } } +func TestProjectJudgeChecksPersistedSourceEpochAcrossGuards(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 }) + epoch := "duty-1" + publication := "pub-1" + items := []extension.ProjectContextItem{{SourceID: "restricted-doc", Revision: "r1", Hash: "h1", Text: "OLD_RESTRICTED_BODY", Required: true}} + m.SetProjectContextResolver(func(_ context.Context, req extension.ProjectContextRequest) (extension.ProjectContextBundle, error) { + return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID, AuthorizationEpoch: epoch, PublicationEpoch: publication, Watermark: "commit-1", Items: items}, nil + }) + rc := &runCtx{mgr: m, run: run, ctx: context.Background()} + started := time.Now().Add(-time.Second) + bundle, _, err := rc.ProjectContext(context.Background(), "write output", 900) + if err != nil { + t.Fatal(err) + } + _, manifest, err := projectcontext.Assemble(bundle, 900) + if err != nil { + t.Fatal(err) + } + 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 { + t.Fatal(err) + } + manifest.Delivery = "responded" + if err := rc.SaveContextManifest(context.Background(), 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 { + t.Fatal(err) + } + if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err != nil { + t.Fatalf("same epoch output rejected: %v", err) + } + if err := rc.ValidateProjectCrossGuard(context.Background(), bundle); err != nil { + t.Fatalf("same epoch cross-guard request rejected: %v", err) + } + if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("replaced bytes")); err == nil { + t.Fatal("replaced output accepted") + } + if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/2", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil { + t.Fatal("second execution reused first attempt's identical output") + } + if err := rc.VerifyProjectStepOutput(context.Background(), "other", "engine-run/other/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil { + t.Fatal("another step reused output binding") + } + items = nil // provider withdraws the document while retaining project membership and epoch + if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil { + t.Fatal("withdrawn document accepted with unchanged authorization epoch") + } + items = []extension.ProjectContextItem{{SourceID: "restricted-doc", Revision: "r1", Hash: "h1", Text: "OLD_RESTRICTED_BODY", Required: true}} + epoch = "duty-2" + current, _, err := rc.ProjectContext(context.Background(), "judge", 900) + if err != nil { + t.Fatal(err) + } + if err := rc.ValidateProjectCrossGuard(context.Background(), current); err == nil { + t.Fatal("cross-guard request ignored persisted epoch change") + } + if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil { + t.Fatal("revoked source epoch accepted for workflow judge") + } + reloaded, err := st.GetRunByID(string(run.ID)) + if err != nil { + t.Fatal(err) + } + resumed := &runCtx{mgr: m, run: reloaded, ctx: context.Background()} + if err := resumed.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil { + t.Fatal("resumed judge ignored revocation") + } + epoch, publication = "duty-1", "pub-2" + if err := resumed.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil { + t.Fatal("resumed judge ignored publication epoch change") + } + other := recoveryRun(t, st, conv, core.RunPending) + noReceipt := &runCtx{mgr: m, run: other, ctx: context.Background()} + if err := noReceipt.BindProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", started, []byte("tool body")); err == nil { + t.Fatal("tool-only output without model provenance accepted") + } + if err := noReceipt.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("tool body")); err == nil { + t.Fatal("another run reused output binding") + } +} + 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 d3648f7..aae73ae 100644 --- a/internal/runtime/project_stream.go +++ b/internal/runtime/project_stream.go @@ -67,6 +67,13 @@ func GuardProjectStream(rc any, goal string, stream agentllm.StreamFunc) agentll if !active { return stream(ctx, model, messages, tools, opts) } + if checker, ok := rc.(interface { + ValidateProjectCrossGuard(context.Context, extension.ProjectContextBundle) error + }); ok { + if err := checker.ValidateProjectCrossGuard(ctx, bundle); err != nil { + return fail(err) + } + } body, manifest, err := projectcontext.Assemble(bundle, budget) if err != nil { return fail(err) @@ -134,6 +141,7 @@ func GuardProjectStream(rc any, goal string, stream agentllm.StreamFunc) agentll return fail(err) } manifest.AttemptID, manifest.Delivery = attemptID, "attempted" + manifest.WorkflowStepAttemptID = projectcontext.WorkflowStepAttempt(ctx) var save func(context.Context, extension.ProjectContextManifest) error if sink, ok := rc.(interface { SaveContextManifest(context.Context, extension.ProjectContextManifest) error diff --git a/internal/runtime/workflow_compiled.go b/internal/runtime/workflow_compiled.go index 428cd90..ac56ec4 100644 --- a/internal/runtime/workflow_compiled.go +++ b/internal/runtime/workflow_compiled.go @@ -64,6 +64,8 @@ func (w CompiledWorkflowRunner) runLaunch(rc RunCtx, launch workflow.Launch, opt } if deps.JudgeStream == nil { deps.JudgeStream = judgeStreamFor(rc) + } else { + deps.JudgeStream = GuardProjectStream(rc, "", deps.JudgeStream) } return wfcompile.RunCompiled(workflowHost{rc: rc}, deps, launch, workflow.Vars{ Input: currentUserText(rc), @@ -169,3 +171,33 @@ func (h workflowHost) ConversationSandbox() string { } return "" } + +func (h workflowHost) InvalidateProjectStepOutput(ctx context.Context, stepID string) error { + invalidate, ok := h.rc.(interface { + InvalidateProjectStepOutput(context.Context, string) error + }) + if !ok { + return fmt.Errorf("project workflow step invalidation unavailable") + } + return invalidate.InvalidateProjectStepOutput(ctx, stepID) +} + +func (h workflowHost) BindProjectStepOutput(ctx context.Context, stepID, stepAttemptID, kind string, startedAt time.Time, output []byte) error { + binder, ok := h.rc.(interface { + BindProjectStepOutput(context.Context, string, string, string, time.Time, []byte) error + }) + if !ok { + return fmt.Errorf("project workflow step binding unavailable") + } + return binder.BindProjectStepOutput(ctx, stepID, stepAttemptID, kind, startedAt, output) +} + +func (h workflowHost) VerifyProjectStepOutput(ctx context.Context, stepID, stepAttemptID, kind string, output []byte) error { + verifier, ok := h.rc.(interface { + VerifyProjectStepOutput(context.Context, string, string, string, []byte) error + }) + if !ok { + return fmt.Errorf("project workflow step verification unavailable") + } + return verifier.VerifyProjectStepOutput(ctx, stepID, stepAttemptID, kind, output) +} diff --git a/internal/runtime/workflow_compiled_test.go b/internal/runtime/workflow_compiled_test.go index 6308895..51b1301 100644 --- a/internal/runtime/workflow_compiled_test.go +++ b/internal/runtime/workflow_compiled_test.go @@ -11,6 +11,7 @@ import ( "testing" agentllm "agent-native/runtime/llm" + "agent-system-backend/internal/projectcontext" "agent-system-backend/internal/workflow" "agent-system-backend/pkg/extension" core "agent-system-core" @@ -296,18 +297,31 @@ func TestJudgeStreamUsesConfiguredModel(t *testing.T) { type projectJudgeCtx struct { *wfFake manifests []extension.ProjectContextManifest + authEpoch string } func (c *projectJudgeCtx) Run() core.Run { return core.Run{ID: "run-judge", Owner: "alice", ProjectID: "project-1"} } func (c *projectJudgeCtx) ProjectContext(context.Context, string, int) (extension.ProjectContextBundle, bool, error) { - return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "project-1", AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{{SourceID: "judge-source", Text: "current project rule", Required: true}}}, true, nil + epoch := c.authEpoch + if epoch == "" { + epoch = "duty-1" + } + return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "project-1", AuthorizationEpoch: epoch, PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{{SourceID: "judge-source", Text: "current project rule", Required: true}}}, true, nil } func (c *projectJudgeCtx) SaveContextManifest(_ context.Context, m extension.ProjectContextManifest) error { c.manifests = append(c.manifests, m) return nil } +func (c *projectJudgeCtx) ValidateProjectCrossGuard(_ context.Context, current extension.ProjectContextBundle) error { + for _, m := range c.manifests { + if m.AuthorizationEpoch != current.AuthorizationEpoch || m.PublicationEpoch != current.PublicationEpoch || m.Watermark != current.Watermark { + return fmt.Errorf("prior workflow model source was revoked") + } + } + return nil +} func TestWorkflowJudgeRequestUsesProjectContext(t *testing.T) { var sent []string @@ -346,4 +360,53 @@ func TestWorkflowJudgeRequestUsesProjectContext(t *testing.T) { } } +func TestWorkflowModelReceiptCarriesEngineStepAttempt(t *testing.T) { + c := &projectJudgeCtx{wfFake: &wfFake{}, authEpoch: "duty-1"} + ctx := projectcontext.WithWorkflowStepAttempt(context.Background(), "engine-run/draft/2") + for ev := range GuardProjectStream(c, "draft", scriptStream(textTurns("done"), nil))(ctx, "model", []agentllm.Message{{Role: agentllm.RoleUser, Content: "write"}}, nil, agentllm.StreamOptions{}) { + if ev.Kind == agentllm.KindError { + t.Fatalf("step model failed: %+v", ev) + } + } + if len(c.manifests) != 2 || c.manifests[0].WorkflowStepAttemptID != "engine-run/draft/2" || c.manifests[1].WorkflowStepAttemptID != "engine-run/draft/2" || c.manifests[1].Delivery != "responded" { + t.Fatalf("engine attempt missing from trusted model receipts: %+v", c.manifests) + } +} + +func TestWorkflowJudgeBlocksOutputFromEarlierAuthorizationEpoch(t *testing.T) { + const old = "OLD_RESTRICTED_BODY" + seen := false + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body struct { + Messages []struct { + Content string `json:"content"` + } `json:"messages"` + } + _ = json.NewDecoder(r.Body).Decode(&body) + for _, m := range body.Messages { + if strings.Contains(m.Content, old) { + seen = true + } + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, "data: [DONE]\n\n") + })) + defer srv.Close() + c := &projectJudgeCtx{wfFake: &wfFake{settings: core.EffectiveSettings{"models": map[string]any{"chat": map[string]any{"baseUrl": srv.URL, "apiKey": "test-key", "model": "test-model"}}}}, authEpoch: "duty-1"} + step := GuardProjectStream(c, "write", scriptStream(textTurns("step done"), nil)) + for range step(context.Background(), "test-model", []agentllm.Message{{Role: agentllm.RoleUser, Content: "write report"}}, nil, agentllm.StreamOptions{}) { + } + if len(c.manifests) != 2 || c.manifests[1].Delivery != "responded" { + t.Fatalf("step source receipt missing: %+v", c.manifests) + } + c.authEpoch = "duty-2" // member remains; document duty and publication changed before critic + blocked := false + for ev := range judgeStreamFor(c)(context.Background(), "test-model", []agentllm.Message{{Role: agentllm.RoleUser, Content: "Judge output.md:\n" + old}}, nil, agentllm.StreamOptions{}) { + blocked = blocked || ev.Kind == agentllm.KindError + } + if seen || !blocked { + t.Fatalf("old workflow output reached a separate judge guard: seen=%v blocked=%v", seen, blocked) + } +} + func (*wfFake) Surfaces() []core.SurfaceSnapshot { return nil } diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 5f26e2c..f5ab653 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -862,6 +862,13 @@ func init() { }, down: map[string][]string{"sqlite": {`DROP INDEX IF EXISTS uniq_context_manifest_terminal`, `DROP INDEX IF EXISTS idx_context_manifests_owner_target`, `DROP TABLE IF EXISTS context_manifests`}, "postgres": {`DROP INDEX IF EXISTS uniq_context_manifest_terminal`, `DROP INDEX IF EXISTS idx_context_manifests_owner_target`, `DROP TABLE IF EXISTS context_manifests`}}, }, + migration{version: 33, name: "workflow_step_sources", + up: map[string][]string{ + "sqlite": {`CREATE TABLE workflow_step_sources(run_id TEXT NOT NULL,step_id TEXT NOT NULL,owner TEXT NOT NULL,project_id TEXT NOT NULL,step_attempt_id TEXT NOT NULL,output_sha256 TEXT NOT NULL,attempt_id TEXT NOT NULL,authorization_epoch TEXT NOT NULL,publication_epoch TEXT NOT NULL,watermark TEXT NOT NULL,version TEXT NOT NULL,created_at TEXT NOT NULL,PRIMARY KEY(run_id,step_id),FOREIGN KEY(run_id) REFERENCES runs(id) ON DELETE CASCADE)`}, + "postgres": {`CREATE TABLE workflow_step_sources(run_id TEXT NOT NULL,step_id TEXT NOT NULL,owner TEXT NOT NULL,project_id TEXT NOT NULL,step_attempt_id TEXT NOT NULL,output_sha256 TEXT NOT NULL,attempt_id TEXT NOT NULL,authorization_epoch TEXT NOT NULL,publication_epoch TEXT NOT NULL,watermark TEXT NOT NULL,version TEXT NOT NULL,created_at TEXT NOT NULL,PRIMARY KEY(run_id,step_id),FOREIGN KEY(run_id) REFERENCES runs(id) ON DELETE CASCADE)`}, + }, + down: map[string][]string{"sqlite": {`DROP TABLE IF EXISTS workflow_step_sources`}, "postgres": {`DROP TABLE IF EXISTS workflow_step_sources`}}, + }, ) } diff --git a/internal/store/migrate_compat.go b/internal/store/migrate_compat.go index 6458fd4..efb7cb2 100644 --- a/internal/store/migrate_compat.go +++ b/internal/store/migrate_compat.go @@ -37,6 +37,7 @@ var migrationBreaksOldReaders = map[int]bool{ 30: false, // project_event_publication_proofs：新表，旧读者不受影响 31: false, // work_task_bindings：新增绑定表，旧后台任务读面不受影响 32: false, // trusted_context_manifests：独立可信回执表，旧读者不受影响 + 33: false, // workflow_step_sources：新增逐步骤来源绑定表，旧读者不受影响 } // SchemaTarget 是本二进制会迁移到的 schema 版本。 diff --git a/internal/store/workflow_step_source.go b/internal/store/workflow_step_source.go new file mode 100644 index 0000000..3ff9efd --- /dev/null +++ b/internal/store/workflow_step_source.go @@ -0,0 +1,128 @@ +package store + +import ( + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "time" + + "agent-system-backend/pkg/extension" +) + +type WorkflowStepSource struct { + RunID, StepID, Owner, ProjectID, StepAttemptID string + OutputSHA256 [32]byte + AttemptID, AuthorizationEpoch, PublicationEpoch, Watermark, Version string +} + +// UnbindWorkflowStepSource removes an earlier attempt before the engine begins +// another execution of the same step. A failed/unknown retry cannot inherit a +// previous output's proof, even if the output bytes happen to be identical. +func (s *Store) UnbindWorkflowStepSource(owner, runID, stepID string) error { + if owner == "" || runID == "" || stepID == "" { + return errors.New("workflow step source scope is incomplete") + } + _, err := s.db.Exec(s.q(`DELETE FROM workflow_step_sources WHERE owner=? AND run_id=? AND step_id=?`), owner, runID, stepID) + return err +} + +// BindWorkflowStepSource associates a materialized prompt step with only the +// responded model receipts created during that step. The caller must have +// excluded parallel prompt execution, whose receipt attribution is ambiguous. +func (s *Store) BindWorkflowStepSource(owner, runID, stepID, stepAttemptID, kind string, startedAt time.Time, output []byte) (WorkflowStepSource, error) { + if kind != "prompt" || owner == "" || runID == "" || stepID == "" || stepAttemptID == "" || startedAt.IsZero() { + return WorkflowStepSource{}, errors.New("workflow step source requires a started prompt step") + } + run, err := s.GetRunByID(runID) + 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) + if err != nil { + return WorkflowStepSource{}, err + } + defer rows.Close() + var latest extension.ProjectContextManifest + var latestAt time.Time + for rows.Next() { + var atRaw, raw string + if err := rows.Scan(&atRaw, &raw); err != nil { + return WorkflowStepSource{}, err + } + at, err := time.Parse(time.RFC3339Nano, atRaw) + if err != nil { + return WorkflowStepSource{}, err + } + if at.Before(startedAt) { + continue + } + var m extension.ProjectContextManifest + if err := json.Unmarshal([]byte(raw), &m); err != nil { + return WorkflowStepSource{}, err + } + if m.WorkflowStepAttemptID != stepAttemptID { + continue + } + 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) { + return WorkflowStepSource{}, errors.New("workflow step model receipts span project authorization revisions") + } + if latestAt.IsZero() || at.After(latestAt) { + latest, latestAt = m, at + } + } + if err := rows.Err(); err != nil { + return WorkflowStepSource{}, err + } + if latestAt.IsZero() { + return WorkflowStepSource{}, errors.New("workflow prompt output has no responded model receipt") + } + b := WorkflowStepSource{RunID: runID, StepID: stepID, Owner: owner, ProjectID: string(run.ProjectID), StepAttemptID: stepAttemptID, OutputSHA256: sha256.Sum256(output), + AttemptID: latest.AttemptID, AuthorizationEpoch: latest.AuthorizationEpoch, PublicationEpoch: latest.PublicationEpoch, Watermark: latest.Watermark, Version: latest.Version} + _, err = s.db.Exec(s.q(`INSERT INTO workflow_step_sources(run_id,step_id,owner,project_id,step_attempt_id,output_sha256,attempt_id,authorization_epoch,publication_epoch,watermark,version,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(run_id,step_id) DO UPDATE SET owner=excluded.owner,project_id=excluded.project_id,step_attempt_id=excluded.step_attempt_id,output_sha256=excluded.output_sha256,attempt_id=excluded.attempt_id,authorization_epoch=excluded.authorization_epoch,publication_epoch=excluded.publication_epoch,watermark=excluded.watermark,version=excluded.version,created_at=excluded.created_at`), + runID, stepID, owner, b.ProjectID, b.StepAttemptID, hex.EncodeToString(b.OutputSHA256[:]), b.AttemptID, b.AuthorizationEpoch, b.PublicationEpoch, b.Watermark, b.Version, nowISO()) + if err != nil { + return WorkflowStepSource{}, fmt.Errorf("bind workflow step output: %w", err) + } + return b, nil +} + +func (s *Store) WorkflowStepSource(owner, runID, stepID string) (WorkflowStepSource, error) { + var b WorkflowStepSource + var hash string + err := s.db.QueryRow(s.q(`SELECT run_id,step_id,owner,project_id,step_attempt_id,output_sha256,attempt_id,authorization_epoch,publication_epoch,watermark,version FROM workflow_step_sources WHERE owner=? AND run_id=? AND step_id=?`), + owner, runID, stepID).Scan(&b.RunID, &b.StepID, &b.Owner, &b.ProjectID, &b.StepAttemptID, &hash, &b.AttemptID, &b.AuthorizationEpoch, &b.PublicationEpoch, &b.Watermark, &b.Version) + if errors.Is(err, sql.ErrNoRows) { + return WorkflowStepSource{}, ErrNotFound + } + if err != nil { + return WorkflowStepSource{}, err + } + decoded, err := hex.DecodeString(hash) + if err != nil || len(decoded) != len(b.OutputSHA256) { + return WorkflowStepSource{}, errors.New("workflow step source has invalid output hash") + } + copy(b.OutputSHA256[:], decoded) + return b, nil +} + +func (s *Store) RespondedContextManifest(owner, runID, attemptID string) (extension.ProjectContextManifest, error) { + var raw string + err := s.db.QueryRow(s.q(`SELECT manifest_json FROM context_manifests WHERE owner=? AND target_id=? AND attempt_id=? AND delivery='responded'`), owner, runID, attemptID).Scan(&raw) + if errors.Is(err, sql.ErrNoRows) { + return extension.ProjectContextManifest{}, ErrNotFound + } + if err != nil { + return extension.ProjectContextManifest{}, err + } + var m extension.ProjectContextManifest + if err := json.Unmarshal([]byte(raw), &m); err != nil { + return extension.ProjectContextManifest{}, err + } + return m, nil +} diff --git a/internal/store/workflow_step_source_test.go b/internal/store/workflow_step_source_test.go new file mode 100644 index 0000000..2c68602 --- /dev/null +++ b/internal/store/workflow_step_source_test.go @@ -0,0 +1,75 @@ +package store + +import ( + "crypto/sha256" + "path/filepath" + "testing" + "time" + + "agent-system-backend/pkg/extension" + core "agent-system-core" +) + +func TestWorkflowStepSourceRequiresRespondedManifestAndExactRunStep(t *testing.T) { + st, err := OpenSQLite(filepath.Join(t.TempDir(), "step.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + p, err := st.CreateProject("alice", "P") + if err != nil { + t.Fatal(err) + } + c, err := st.CreateConversation("alice", string(p.ID), "C", "") + if err != nil { + t.Fatal(err) + } + r := core.Run{Owner: "alice", ConversationID: c.ID, ProjectID: p.ID, Kind: core.RunConversation, Status: core.RunPending} + if err := st.CreateRun(&r); err != nil { + t.Fatal(err) + } + 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 { + 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("unanswered model attempt bound output") + } + unrelated := m + unrelated.AttemptID = "other-model-attempt" + unrelated.WorkflowStepAttemptID = "engine-run/other/1" + if err := st.AppendContextManifest("alice", string(r.ID), unrelated); err != nil { + t.Fatal(err) + } + unrelated.Delivery = "responded" + if err := st.AppendContextManifest("alice", string(r.ID), unrelated); 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 { + t.Fatal(err) + } + bound, err := st.BindWorkflowStepSource("alice", string(r.ID), "draft", "engine-run/draft/1", "prompt", started, []byte("old body")) + if err != nil || bound.AttemptID != "attempt-1" || bound.StepAttemptID != "engine-run/draft/1" || bound.OutputSHA256 != sha256.Sum256([]byte("old body")) { + t.Fatalf("source binding missed actual output: %+v err=%v", bound, err) + } + if _, err := st.WorkflowStepSource("alice", string(r.ID), "other"); err == nil { + t.Fatal("another step reused output binding") + } + if _, err := st.WorkflowStepSource("bob", string(r.ID), "draft"); err == nil { + t.Fatal("another owner read output binding") + } + 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") + } + if err := st.UnbindWorkflowStepSource("alice", string(r.ID), "draft"); err != nil { + t.Fatal(err) + } + if _, err := st.WorkflowStepSource("alice", string(r.ID), "draft"); err == nil { + t.Fatal("retry retained an earlier source binding") + } +} diff --git a/internal/wfcompile/host.go b/internal/wfcompile/host.go index 67434b6..b0ca13a 100644 --- a/internal/wfcompile/host.go +++ b/internal/wfcompile/host.go @@ -36,6 +36,15 @@ type Host interface { Tools() []core.ToolDesc } +// ProjectStepProvenance is implemented by the trusted Backend run host. Prompt +// output is bound after materialization; a critic must verify the exact bytes +// and current project authorization before it includes output in a model call. +type ProjectStepProvenance interface { + InvalidateProjectStepOutput(context.Context, string) error + BindProjectStepOutput(context.Context, string, string, string, time.Time, []byte) error + VerifyProjectStepOutput(context.Context, string, string, string, []byte) error +} + // Spawner 可选子代理派发能力（脚本回调 agent.spawn 用）。 type Spawner interface { SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) @@ -74,10 +83,11 @@ const ( // hostRunner 宿主 intent 步执行器（squad.IntentRunner 实现）。 type hostRunner struct { - rc Host - deps Deps - scripts map[string]string - subOf map[string]string + rc Host + deps Deps + scripts map[string]string + subOf map[string]string + stepKinds map[string]string mu sync.Mutex verdicts map[string]string // verdict 字段 → 最近一次判定 @@ -421,6 +431,20 @@ func (r *hostRunner) doCritic(ctx context.Context, step squad.IntentStep) (strin target := step.Slots["target_step"] field := step.Slots["verdict_field"] output := r.readStepOutput(step.WorkDir, target) + if r.rc.Run().ProjectID != "" { + guard, ok := r.rc.(ProjectStepProvenance) + if !ok { + return "", nil, fmt.Errorf("项目工作流产物缺少可信来源校验") + } + wsDir := filepath.Dir(filepath.Dir(step.WorkDir)) + stepAttemptID, err := completedStepAttempt(wsDir, target) + if err != nil { + return "", nil, fmt.Errorf("项目工作流步骤 %s 状态不可核验: %w", target, err) + } + if err := guard.VerifyProjectStepOutput(ctx, target, stepAttemptID, r.stepKinds[target], []byte(output)); err != nil { + return "", nil, fmt.Errorf("项目工作流步骤 %s 产物依据不可用: %w", target, err) + } + } verdict := r.judge(ctx, step.Slots["goal"], step.Slots["criteria"], output) fields := map[string]string{} if verdict.Passed { diff --git a/internal/wfcompile/project_backend.go b/internal/wfcompile/project_backend.go new file mode 100644 index 0000000..47e2e1c --- /dev/null +++ b/internal/wfcompile/project_backend.go @@ -0,0 +1,37 @@ +package wfcompile + +import ( + "context" + + "agent-squad/squad" + "agent-system-backend/internal/projectcontext" +) + +// projectStepBackends carries the engine-generated execution ID through the +// executor's request context into each model manifest. This associates a +// responded model request with exactly one workflow step attempt. +func projectStepBackends(base *squad.BackendResolver) *squad.BackendResolver { + wrapped := squad.NewBackendResolver() + for _, name := range base.ListBackends() { + backendName := name + wrapped.Register(backendName, func(role squad.RoleSpec) (squad.Executor, error) { + backend, _, err := base.Resolve(role, backendName) + if err != nil { + return nil, err + } + return &projectStepExecutor{next: backend}, nil + }) + } + return wrapped +} + +type projectStepExecutor struct{ next squad.Executor } + +func (e *projectStepExecutor) Name() string { return e.next.Name() } + +func (e *projectStepExecutor) Execute(ctx context.Context, req squad.ExecRequest) (*squad.ExecResult, error) { + if req.ExecutionID == "" { + return &squad.ExecResult{Success: false, ErrorReason: "project workflow step attempt unavailable"}, nil + } + return e.next.Execute(projectcontext.WithWorkflowStepAttempt(ctx, req.ExecutionID), req) +} diff --git a/internal/wfcompile/project_sources.go b/internal/wfcompile/project_sources.go new file mode 100644 index 0000000..0c971ab --- /dev/null +++ b/internal/wfcompile/project_sources.go @@ -0,0 +1,107 @@ +package wfcompile + +import ( + "context" + "errors" + "os" + "path/filepath" + "sync" + "time" + + "agent-squad/squad" +) + +// completedStepAttempt comes from the squad engine's persisted state, not a +// model response or plugin argument. It identifies the exact execution that +// produced output.md, including after a workflow resumes. +func completedStepAttempt(wsDir, stepID string) (string, error) { + state, err := (squad.FileStore{}).LoadRun(wsDir) + if err != nil { + return "", err + } + for _, step := range state.Plan.Workflow.Steps { + if step.ID == stepID && step.Status == "completed" && step.AttemptID != "" { + return step.AttemptID, nil + } + } + return "", errors.New("completed step attempt unavailable") +} + +// projectOutputSink binds only a completed prompt step's materialized bytes. +// Parallel prompt executions have no unambiguous request-to-step attribution +// in the current backend contract, so they receive no binding and the critic +// fails closed. A recovered completion without its start event likewise needs +// an existing persisted binding to pass its critic. +type projectOutputSink struct { + rc ProjectStepProvenance + ctx context.Context + root string + subOf map[string]string + kinds map[string]string + mu sync.Mutex + started map[string]time.Time + ambiguous map[string]bool +} + +func newProjectOutputSink(rc Host, root string, subOf, kinds map[string]string) squad.ProgressSink { + binder := rc.(ProjectStepProvenance) + return &projectOutputSink{rc: binder, ctx: rc.Context(), root: root, subOf: subOf, kinds: kinds, + started: map[string]time.Time{}, ambiguous: map[string]bool{}} +} + +func (s *projectOutputSink) Emit(ev squad.Event) { + if ev.StepID == "" { + return + } + s.mu.Lock() + switch ev.Type { + case squad.EvStepStarted: + if s.kinds[ev.StepID] == "prompt" { + if err := s.rc.InvalidateProjectStepOutput(s.ctx, ev.StepID); err != nil { + s.ambiguous[ev.StepID] = true + } + } + if len(s.started) != 0 { + if s.kinds[ev.StepID] == "prompt" { + s.ambiguous[ev.StepID] = true + } + for id := range s.started { + if s.kinds[id] == "prompt" { + s.ambiguous[id] = true + } + } + } + s.started[ev.StepID] = ev.At + s.mu.Unlock() + return + case squad.EvStepCompleted: + startedAt, started := s.started[ev.StepID] + ambiguous := s.ambiguous[ev.StepID] + delete(s.started, ev.StepID) + delete(s.ambiguous, ev.StepID) + s.mu.Unlock() + if s.kinds[ev.StepID] != "prompt" || !started || startedAt.IsZero() || ambiguous { + return + } + sub := s.subOf[ev.StepID] + if sub == "" { + return + } + data, err := os.ReadFile(filepath.Join(s.root, "runs", ev.RunID, filepath.FromSlash(sub), "output.md")) + if err == nil { + wsDir := filepath.Join(s.root, "runs", ev.RunID) + attemptID, attemptErr := completedStepAttempt(wsDir, ev.StepID) + if attemptErr == nil { + _ = s.rc.BindProjectStepOutput(s.ctx, ev.StepID, attemptID, "prompt", startedAt, data) + } + } + return + case squad.EvStepFailed: + delete(s.started, ev.StepID) + delete(s.ambiguous, ev.StepID) + s.mu.Unlock() + return + default: + s.mu.Unlock() + } +} diff --git a/internal/wfcompile/run.go b/internal/wfcompile/run.go index 661c5bf..8799e13 100644 --- a/internal/wfcompile/run.go +++ b/internal/wfcompile/run.go @@ -76,6 +76,15 @@ func RunCompiled(rc Host, deps Deps, launch workflow.Launch, vars workflow.Vars, if err != nil { return Outcome{}, err } + stepKinds := make(map[string]string, len(def.Steps)) + for _, step := range def.Steps { + stepKinds[step.ID] = step.Kind + } + if rc.Run().ProjectID != "" { + if _, ok := rc.(ProjectStepProvenance); !ok { + return Outcome{}, fmt.Errorf("项目工作流缺少可信步骤产物绑定能力") + } + } registry := squad.NewRoleRegistry() for _, role := range compiled.Plan.Squad.Agents { @@ -87,9 +96,17 @@ func RunCompiled(rc Host, deps Deps, launch workflow.Launch, vars workflow.Vars, if backends == nil { backends = squad.NewBackendResolver() } + if rc.Run().ProjectID != "" { + backends = projectStepBackends(backends) + } runner := &hostRunner{ rc: rc, deps: deps, scripts: launch.Scripts, subOf: compiled.SubOf, - verdicts: map[string]string{}, lastWhy: map[string]string{}, fails: map[string]int{}, + stepKinds: stepKinds, + verdicts: map[string]string{}, lastWhy: map[string]string{}, fails: map[string]int{}, + } + var progress squad.ProgressSink = squad.NopSink{} + if rc.Run().ProjectID != "" { + progress = newProjectOutputSink(rc, filepath.Join(deps.DataDir, "wf", string(rc.Run().ID)), compiled.SubOf, stepKinds) } engine, err := squad.NewEngine(squad.Options{ DataDir: filepath.Join(deps.DataDir, "wf", string(rc.Run().ID)), @@ -102,7 +119,7 @@ func RunCompiled(rc Host, deps Deps, launch workflow.Launch, vars workflow.Vars, Checker: passChecker{}, IntentCatalog: compiled.Catalog, IntentRunner: runner, - Sink: squad.NopSink{}, + Sink: progress, }) if err != nil { return Outcome{}, fmt.Errorf("workflow: 引擎装配: %w", err) diff --git a/internal/wfcompile/wfcompile_test.go b/internal/wfcompile/wfcompile_test.go index 6881c70..a9a57e2 100644 --- a/internal/wfcompile/wfcompile_test.go +++ b/internal/wfcompile/wfcompile_test.go @@ -2,15 +2,19 @@ package wfcompile import ( "context" + "crypto/sha256" "encoding/json" "fmt" "os" + "path/filepath" "strings" "sync" "testing" + "time" agentllm "agent-native/runtime/llm" "agent-squad/squad" + "agent-system-backend/internal/projectcontext" "agent-system-backend/internal/workflow" core "agent-system-core" ) @@ -452,6 +456,118 @@ func TestRunCompiledDemoBrief(t *testing.T) { } } +type projectSourceHost struct { + *fakeHost + boundID string + boundHash [32]byte + verified int + invalidated int +} + +type stepContextProbe struct{ got string } + +func (p *stepContextProbe) Name() string { return "probe" } +func (p *stepContextProbe) Execute(ctx context.Context, _ squad.ExecRequest) (*squad.ExecResult, error) { + p.got = projectcontext.WorkflowStepAttempt(ctx) + return &squad.ExecResult{Success: true}, nil +} + +func TestProjectStepBackendCarriesEngineAttempt(t *testing.T) { + probe := &stepContextProbe{} + base := squad.NewBackendResolver() + base.RegisterInstance("probe", probe) + backend, _, err := projectStepBackends(base).Resolve(squad.RoleSpec{Role: "writer", Backend: "probe"}, "") + if err != nil { + t.Fatal(err) + } + if _, err := backend.Execute(context.Background(), squad.ExecRequest{ExecutionID: "engine-run/draft/2"}); err != nil { + t.Fatal(err) + } + if probe.got != "engine-run/draft/2" { + t.Fatalf("model backend lost trusted step attempt: %q", probe.got) + } +} + +func (h *projectSourceHost) Run() core.Run { + return core.Run{ID: "run_project_wf", ProjectID: "p1", Owner: "alice"} +} +func (h *projectSourceHost) BindProjectStepOutput(_ context.Context, stepID, stepAttemptID, kind string, startedAt time.Time, output []byte) error { + if kind != "prompt" || stepAttemptID == "" || startedAt.IsZero() { + return fmt.Errorf("untrusted step binding: %s %s %s", stepID, kind, startedAt) + } + h.boundID, h.boundHash = stepID, sha256.Sum256(output) + return nil +} +func (h *projectSourceHost) VerifyProjectStepOutput(_ context.Context, stepID, stepAttemptID, kind string, output []byte) error { + h.verified++ + if stepID != h.boundID || stepAttemptID == "" || kind != "prompt" || sha256.Sum256(output) != h.boundHash { + return fmt.Errorf("step output provenance mismatch: bound=%q/%x got=%q/%x", h.boundID, h.boundHash, stepID, sha256.Sum256(output)) + } + return nil +} +func (h *projectSourceHost) InvalidateProjectStepOutput(_ context.Context, stepID string) error { + if stepID == h.boundID { + h.boundID = "" + h.boundHash = [32]byte{} + } + h.invalidated++ + return nil +} + +func TestProjectOutputSinkRejectsConcurrentAndRetriedAttribution(t *testing.T) { + h := &projectSourceHost{fakeHost: &fakeHost{}} + root := t.TempDir() + sub := "tasks/01_draft" + path := filepath.Join(root, "runs", "engine-run", sub, "output.md") + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("model body"), 0o600); err != nil { + t.Fatal(err) + } + sink := newProjectOutputSink(h, root, map[string]string{"draft": sub}, map[string]string{"draft": "prompt", "other": "tool"}) + started := time.Now().Add(-time.Second) + if err := h.BindProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", started, []byte("old attempt")); err != nil { + t.Fatal(err) + } + sink.Emit(squad.Event{Type: squad.EvStepStarted, RunID: "engine-run", StepID: "draft", At: started}) + if err := h.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("old attempt")); err == nil { + t.Fatal("retry retained previous attempt's binding") + } + sink.Emit(squad.Event{Type: squad.EvStepStarted, RunID: "engine-run", StepID: "other", At: started.Add(time.Millisecond)}) + sink.Emit(squad.Event{Type: squad.EvStepCompleted, RunID: "engine-run", StepID: "draft", At: time.Now()}) + if h.boundID != "" || h.invalidated == 0 { + t.Fatalf("ambiguous retry rebound output: id=%q invalidated=%d", h.boundID, h.invalidated) + } +} + +func TestProjectPromptCriticBindsActualStepOutputBeforeJudge(t *testing.T) { + f := &projectSourceHost{fakeHost: &fakeHost{hitl: []core.HITLResponse{{ChoiceID: "short"}, {ChoiceID: "yes"}}}} + be := &seqBackend{texts: []string{"draft text\n核对标记：P4"}} + _, err := RunCompiled(f, Deps{DataDir: t.TempDir(), Backends: seqResolver(t, be), DefaultBackend: "seq", JudgeStream: verdictStream(verdictPass)}, + workflow.Launch{Def: parseDef(t, demoBriefJSON)}, workflow.Vars{Input: "write", Slots: map[string]string{"topic": "topic"}}, RunOptions{}) + if err != nil { + t.Fatal(err) + } + if f.boundID != "draft" || f.boundHash != sha256.Sum256([]byte("draft text\n核对标记：P4\n")) || f.verified == 0 { + t.Fatalf("project workflow output lacked trusted step binding: id=%q verified=%d", f.boundID, f.verified) + } +} + +func TestProjectToolCriticRejectsUnprovenOutputBeforeModel(t *testing.T) { + f := &projectSourceHost{fakeHost: &fakeHost{toolNames: []string{"time_now"}}} + def := core.WorkflowDef{ID: "tool-critic", Title: "Tool review", Steps: []core.WorkflowStep{{ID: "get", Title: "Read", Kind: "tool", Tool: "time_now", Critic: &core.StepCritic{Criteria: "check"}}}} + called := false + judge := func(ctx context.Context, model string, messages []agentllm.Message, tools []agentllm.ToolDecl, opts agentllm.StreamOptions) <-chan agentllm.Event { + called = true + return verdictStream(verdictPass)(ctx, model, messages, tools, opts) + } + _, err := RunCompiled(f, Deps{DataDir: t.TempDir(), JudgeStream: judge}, workflow.Launch{Def: def}, workflow.Vars{Input: "read"}, RunOptions{}) + if err == nil || called || !strings.Contains(err.Error(), "产物依据不可用") { + t.Fatalf("unproven tool output reached judge: err=%v called=%v", err, called) + } +} + // critic 预算耗尽仍不通过 → 工作流失败（P4 fail-closed）。 func TestRunCompiledCriticExhaustedFails(t *testing.T) { def := parseDef(t, demoBriefJSON) diff --git a/pkg/extension/project_context.go b/pkg/extension/project_context.go index ad85007..2010b0c 100644 --- a/pkg/extension/project_context.go +++ b/pkg/extension/project_context.go @@ -112,22 +112,23 @@ type ProjectContextManifestEntry struct { // ProjectContextManifest describes the text actually passed to the completion // client. Candidate references in a bundle are never listed as included input. type ProjectContextManifest struct { - ProjectID core.ID - TaskID string - Version string - Watermark string - AuthorizationEpoch string - PublicationEpoch string - AttemptID string - Delivery string // attempted | responded | unknown; never implies user-visible delivery - Model string - BudgetChars int - InputSHA256 string - Included []ProjectContextManifestEntry - Dropped []ProjectContextManifestEntry - OfferedReferences []ProjectEvidenceRef - ReadEvidence []ProjectContextManifestEntry - Missing []string + ProjectID core.ID + TaskID string + Version string + Watermark string + AuthorizationEpoch string + PublicationEpoch string + AttemptID string + WorkflowStepAttemptID string + Delivery string // attempted | responded | unknown; never implies user-visible delivery + Model string + BudgetChars int + InputSHA256 string + Included []ProjectContextManifestEntry + Dropped []ProjectContextManifestEntry + OfferedReferences []ProjectEvidenceRef + ReadEvidence []ProjectContextManifestEntry + Missing []string } type WorkCompletionResult struct {