diff --git a/cmd/storage-migrate/integration_test.go b/cmd/storage-migrate/integration_test.go index d4997c0..5fa87c9 100644 --- a/cmd/storage-migrate/integration_test.go +++ b/cmd/storage-migrate/integration_test.go @@ -9,8 +9,51 @@ import ( "testing" "agent-system-backend/internal/store" + "agent-system-backend/pkg/extension" ) +func TestMigrateTrustedContextManifestWithoutRawScope(t *testing.T) { + target, cleanup := pgSchemaDSN(t) + defer cleanup() + fixture := buildFixture(t) + src, err := store.OpenSQLite(fixture) + if err != nil { + t.Fatal(err) + } + m := extension.ProjectContextManifest{ProjectID: "p1", AttemptID: "model-attempt", InputSHA256: "request-hash", Delivery: "attempted", + Included: []extension.ProjectContextManifestEntry{{SourceID: "source-a", Revision: "r1", SourceHash: "ha", InputSHA256: "input-hash"}}} + if err := src.AppendContextManifestWithScope("alice", "r1", m, store.ContextManifestScope{Bound: true}); err != nil { + t.Fatal(err) + } + m.Delivery = "responded" + if err := src.AppendContextManifestWithScope("alice", "r1", m, store.ContextManifestScope{Bound: true}); err != nil { + t.Fatal(err) + } + if err := src.Close(); err != nil { + t.Fatal(err) + } + snapshot := filepath.Join(t.TempDir(), "snapshot.db") + if err := cmdBackup([]string{"--source", fixture, "--out", snapshot}); err != nil { + t.Fatal(err) + } + if err := cmdMigrate([]string{"--source", snapshot, "--target", target, "--migration-id", "context-copy"}); err != nil { + t.Fatal(err) + } + pg, err := store.OpenPostgres(target, filepath.Join(t.TempDir(), "artifacts")) + if err != nil { + t.Fatal(err) + } + defer pg.Close() + receipts, err := pg.ListContextManifests("alice", "r1") + if err != nil || len(receipts) != 2 || len(receipts[1].Included) != 1 || receipts[1].Included[0].SourceID != "source-a" { + t.Fatalf("content-free model receipts not copied: count=%d err=%v", len(receipts), err) + } + var rawColumns int + if err := pg.DB().QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='context_manifests' AND column_name IN ('resolve_goal','resolve_target_id')`).Scan(&rawColumns); err != nil || rawColumns != 0 { + t.Fatalf("migrated PG schema has raw prompt columns: count=%d err=%v", rawColumns, err) + } +} + // U4 集成(环境门控 AGENT_PG_TEST_DSN:CI test-pg job / 本机隔离实例): // 夹具 → 受管迁移 → 校验和验证 → 幂等重跑 → 断点续传 → 非空目标拒绝 → // 未知 owner 阻塞。每用例独立 PG schema(与 store 测试同隔离约定)。 diff --git a/cmd/storage-migrate/main.go b/cmd/storage-migrate/main.go index 92de9f2..93d9463 100644 --- a/cmd/storage-migrate/main.go +++ b/cmd/storage-migrate/main.go @@ -242,6 +242,9 @@ func cmdBackup(args []string) error { return err } defer db.Close() + if err := refuseCandidateRawContextScope(db); err != nil { + return err + } if _, err := db.Exec(`VACUUM INTO ?`, *out); err != nil { return fmt.Errorf("vacuum into: %w", err) } @@ -336,6 +339,9 @@ func cmdMigrate(args []string) error { return err } defer src.Close() + if err := refuseCandidateRawContextScope(src); err != nil { + return err + } snapID, _, err := fileSHA256(*source) if err != nil { return err @@ -410,6 +416,23 @@ func cmdMigrate(args []string) error { return writeReport(rep, *out) } +// Candidate W6 v34 may contain full model prompts in private scope columns. +// Never copy or back up such a source implicitly; it must be migrated and +// reviewed in an isolated Backend store before this tool accepts it. +func refuseCandidateRawContextScope(db *sql.DB) error { + if !sqliteHasTable(db, "context_manifests") { + return nil + } + var columns int + if err := db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('context_manifests') WHERE name IN ('resolve_goal','resolve_target_id')`).Scan(&columns); err != nil { + return err + } + if columns > 0 { + return fmt.Errorf("candidate context scope text columns remain; isolate and scrub source before backup or migration") + } + return nil +} + // guardIdentity:目标非空且身份不同即拒绝(R7:不覆盖他人数据)。 // - 存在其他 migration_id 的完成记录 → 拒绝; // - 同 migration_id 但 source_snapshot_id 不同(换了快照续传)→ 拒绝; diff --git a/cmd/storage-migrate/main_test.go b/cmd/storage-migrate/main_test.go index 8731e0e..24b9170 100644 --- a/cmd/storage-migrate/main_test.go +++ b/cmd/storage-migrate/main_test.go @@ -1,7 +1,10 @@ package main import ( + "bytes" + "database/sql" "encoding/json" + "os" "path/filepath" "strings" "testing" @@ -195,3 +198,53 @@ func TestCopyOrderCoversAllFixtureTables(t *testing.T) { t.Fatalf("copyOrder 漏表(新表加入 schema 后须同步工具清单):%v", missing) } } + +func TestBackupAndMigrateRejectCandidateRawContextScope(t *testing.T) { + path := buildFixture(t) + db, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`ALTER TABLE context_manifests ADD COLUMN resolve_goal TEXT NOT NULL DEFAULT ''`); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`ALTER TABLE context_manifests ADD COLUMN resolve_target_id TEXT NOT NULL DEFAULT ''`); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`DELETE FROM schema_migrations WHERE version=35`); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`INSERT INTO context_manifests(id,attempt_id,owner,target_id,project_id,input_sha256,delivery,manifest_json,created_at,scope_bound,resolve_goal,resolve_target_id) VALUES('candidate-ctx','candidate-attempt','alice','r1','p1','input-hash','attempted','{}','2026-10-14T00:00:00Z',1,'OLD_RESTRICTED_BODY','doc-private')`); err != nil { + t.Fatal(err) + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + out := filepath.Join(t.TempDir(), "unsafe-backup.db") + if err := cmdBackup([]string{"--source", path, "--out", out}); err == nil || !strings.Contains(err.Error(), "candidate context scope text") { + t.Fatalf("backup accepted candidate v34 raw scope: %v", err) + } + if _, err := os.Stat(out); !os.IsNotExist(err) { + t.Fatalf("rejected backup still created output: %v", err) + } + if err := cmdMigrate([]string{"--source", path, "--target", "postgres://invalid", "--migration-id", "unsafe"}); err == nil || !strings.Contains(err.Error(), "candidate context scope text") { + t.Fatalf("migration accepted candidate v34 raw source: %v", err) + } + scrubbed, err := store.OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + if err := scrubbed.Close(); err != nil { + t.Fatal(err) + } + if err := cmdBackup([]string{"--source", path, "--out", out}); err != nil { + t.Fatalf("scrubbed source rejected: %v", err) + } + data, err := os.ReadFile(out) + if err != nil { + t.Fatal(err) + } + if bytes.Contains(data, []byte("OLD_RESTRICTED_BODY")) { + t.Fatal("backup retained candidate raw scope body") + } +} diff --git a/internal/app/app.go b/internal/app/app.go index 0f9a111..af7521e 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -620,6 +620,7 @@ func Run(opt Options) error { if contextSlot.active() { mgr.SetProjectContextResolver(contextSlot.resolve) mgr.SetProjectEvidenceReader(contextSlot.readEvidence) + mgr.SetProjectContextSourceRechecker(contextSlot.recheck) } mgr.SetKnowledgeProjectScope(func(ctx context.Context, actor string, id core.ID) (core.ID, error) { return projectSlot.resolve(ctx, caps, actor, id) diff --git a/internal/app/project_context.go b/internal/app/project_context.go index fb056d3..1f1aeeb 100644 --- a/internal/app/project_context.go +++ b/internal/app/project_context.go @@ -118,6 +118,33 @@ func (s *projectContextSlot) readEvidence(ctx context.Context, req extension.Pro return evidence, nil } +func (s *projectContextSlot) recheck(ctx context.Context, req extension.ProjectContextSourceRecheckRequest) (extension.ProjectContextSourceRecheckResult, error) { + if s == nil || req.Scope.Actor == "" || req.Scope.ProjectID == "" || req.Scope.Goal != "" { + return extension.ProjectContextSourceRecheckResult{}, errors.New("extension: trusted source recheck scope missing") + } + s.mu.RLock() + p, holder := s.provider, s.holder + s.mu.RUnlock() + rechecker, ok := p.(extension.ProjectContextSourceRechecker) + if !ok { + return extension.ProjectContextSourceRecheckResult{}, errors.New("extension: project source rechecker unavailable") + } + req.Scope.PluginID = holder + bounded, cancel := context.WithTimeout(ctx, 3*time.Second) + defer cancel() + result, err := rechecker.Recheck(bounded, req) + if err != nil { + return extension.ProjectContextSourceRecheckResult{}, err + } + if err := bounded.Err(); err != nil { + return extension.ProjectContextSourceRecheckResult{}, err + } + if result.Bundle.ProjectID != req.Scope.ProjectID || result.Bundle.Status != extension.ProjectContextReady { + return extension.ProjectContextSourceRecheckResult{}, errors.New("extension: project sources no longer authorized") + } + return result, nil +} + const workContextBudgetChars = 12000 // TaskCreateForWork fixes the project binding while the authenticated command diff --git a/internal/app/project_context_test.go b/internal/app/project_context_test.go index 4b40b36..0725ff8 100644 --- a/internal/app/project_context_test.go +++ b/internal/app/project_context_test.go @@ -22,6 +22,46 @@ type testProjectContextProvider struct { read func(extension.ProjectEvidenceRequest) (extension.ProjectEvidence, error) } +type testSourceRecheckProvider struct { + testProjectContextProvider + called *bool +} + +func (p testSourceRecheckProvider) Recheck(ctx context.Context, req extension.ProjectContextSourceRecheckRequest) (extension.ProjectContextSourceRecheckResult, error) { + if _, ok := ctx.Deadline(); !ok || req.Scope.PluginID != "rdm" || req.Scope.Actor != "alice" || req.Scope.ProjectID != "p1" || req.Scope.Goal != "" { + return extension.ProjectContextSourceRecheckResult{}, fmt.Errorf("untrusted or unbounded recheck scope") + } + *p.called = true + return extension.ProjectContextSourceRecheckResult{Bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.Scope.ProjectID}}, nil +} + +func TestProjectSourceRecheckUsesSameRegisteredProviderAndDeadline(t *testing.T) { + called := false + slot := &projectContextSlot{} + provider := testSourceRecheckProvider{called: &called} + if err := (pluginRegistry{pluginID: "rdm", caps: hostcaps{contexts: slot}}).RegisterProjectContextProvider(provider); err != nil { + t.Fatal(err) + } + if _, err := slot.recheck(context.Background(), extension.ProjectContextSourceRecheckRequest{Scope: extension.ProjectContextRequest{Actor: "alice", ProjectID: "p1"}}); err != nil || !called { + t.Fatalf("registered source rechecker unavailable: called=%v err=%v", called, err) + } + if _, err := slot.recheck(context.Background(), extension.ProjectContextSourceRecheckRequest{Scope: extension.ProjectContextRequest{Actor: "alice", ProjectID: "p1", Goal: "OLD_RESTRICTED_BODY"}}); err == nil { + t.Fatal("raw Goal entered source recheck provider") + } + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := slot.recheck(cancelled, extension.ProjectContextSourceRecheckRequest{Scope: extension.ProjectContextRequest{Actor: "alice", ProjectID: "p1"}}); err == nil { + t.Fatal("cancelled source recheck returned authorized context") + } + legacy := &projectContextSlot{} + if err := (pluginRegistry{pluginID: "rdm", caps: hostcaps{contexts: legacy}}).RegisterProjectContextProvider(testProjectContextProvider{}); err != nil { + t.Fatal(err) + } + if _, err := legacy.recheck(context.Background(), extension.ProjectContextSourceRecheckRequest{Scope: extension.ProjectContextRequest{Actor: "alice", ProjectID: "p1"}}); err == nil { + t.Fatal("provider without optional rechecker silently accepted old sources") + } +} + type deadlineContextProvider struct{ sawDeadline *bool } func (p deadlineContextProvider) Resolve(ctx context.Context, req extension.ProjectContextRequest) (extension.ProjectContextBundle, error) { @@ -266,8 +306,8 @@ func TestCompleteForWorkUsesStoredScopeAndActualInputManifest(t *testing.T) { func TestWorkContextBudgetKeepsRequiredEvidenceFirst(t *testing.T) { bundle := extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "project-1", Items: []extension.ProjectContextItem{ - {SourceID: "optional", Text: "12345678"}, - {SourceID: "required", Text: "must keep", Required: true}, + {SourceID: "optional", Revision: "r1", Hash: "h1", Text: "12345678"}, + {SourceID: "required", Revision: "r1", Hash: "h2", Text: "must keep", Required: true}, }} prompt, manifest, err := assembleWorkCompletion("goal", bundle, 140) if err != nil { diff --git a/internal/projectcontext/context.go b/internal/projectcontext/context.go index a8b47d2..255c799 100644 --- a/internal/projectcontext/context.go +++ b/internal/projectcontext/context.go @@ -27,10 +27,16 @@ func Assemble(bundle extension.ProjectContextBundle, budget int) (string, extens missingRequired := false appendItem := func(item extension.ProjectContextItem) { entry := extension.ProjectContextManifestEntry{SourceID: item.SourceID, Revision: item.Revision, SourceHash: item.Hash} + if item.SourceID == "" || item.Revision == "" || item.Hash == "" { + entry.Reason = "unverifiable-source" + m.Dropped = append(m.Dropped, entry) + missingRequired = missingRequired || item.Required + return + } runes := []rune(item.Text) label := "\n[Source " + item.SourceID + " revision " + item.Revision + "]\n" cost := len(runes) + len([]rune(label)) - if cost > remaining || item.SourceID == "" { + if cost > remaining { entry.Reason = "over-budget-or-missing-source" m.Dropped = append(m.Dropped, entry) missingRequired = missingRequired || item.Required @@ -66,7 +72,7 @@ func Assemble(bundle extension.ProjectContextBundle, budget int) (string, extens } } for _, ref := range bundle.Evidence { - if ref.ID == "" || ref.SourceID == "" { + if ref.ID == "" || ref.SourceID == "" || ref.Revision == "" || ref.Hash == "" { continue } line := fmt.Sprintf("\n[Evidence ref %q source %q revision %q]\n", ref.ID, ref.SourceID, ref.Revision) diff --git a/internal/projectcontext/context_test.go b/internal/projectcontext/context_test.go index bf26a3a..abaeda3 100644 --- a/internal/projectcontext/context_test.go +++ b/internal/projectcontext/context_test.go @@ -8,7 +8,7 @@ import ( ) func TestAssembleOffersOnlyBudgetedEvidenceReferences(t *testing.T) { - bundle := extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", Evidence: []extension.ProjectEvidenceRef{{ID: "ref-1", SourceID: "doc-1", Revision: "r2"}, {ID: strings.Repeat("x", 1000), SourceID: "doc-2"}}} + bundle := extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", Evidence: []extension.ProjectEvidenceRef{{ID: "ref-1", SourceID: "doc-1", Revision: "r2", Hash: "h1"}, {ID: strings.Repeat("x", 1000), SourceID: "doc-2", Revision: "r2", Hash: "h2"}}} text, manifest, err := Assemble(bundle, 180) if err != nil { t.Fatal(err) @@ -19,7 +19,7 @@ func TestAssembleOffersOnlyBudgetedEvidenceReferences(t *testing.T) { } func TestValidateCurrentRejectsWithdrawnOfferedReference(t *testing.T) { - ref := extension.ProjectEvidenceRef{ID: "ref-1", SourceID: "doc-1", Revision: "r2"} + ref := extension.ProjectEvidenceRef{ID: "ref-1", SourceID: "doc-1", Revision: "r2", Hash: "h1"} old := extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", Evidence: []extension.ProjectEvidenceRef{ref}} _, manifest, err := Assemble(old, 180) if err != nil { @@ -41,3 +41,10 @@ func TestAssembleIncludesUnresolvedMissingStatus(t *testing.T) { t.Fatalf("missing status hidden from model or receipt: text=%q manifest=%+v", text, manifest) } } + +func TestAssembleRejectsRequiredSourceWithoutStableRevisionAndHash(t *testing.T) { + bundle := extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", Items: []extension.ProjectContextItem{{SourceID: "legacy", Text: "private", Required: true}}} + if _, _, err := Assemble(bundle, 900); err == nil { + t.Fatal("legacy source without revision/hash entered recoverable project model input") + } +} diff --git a/internal/runmgr/project_context.go b/internal/runmgr/project_context.go index 763eb84..11c02b8 100644 --- a/internal/runmgr/project_context.go +++ b/internal/runmgr/project_context.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "errors" "fmt" + "slices" "strings" "time" @@ -26,6 +27,10 @@ func (m *Manager) SetProjectEvidenceReader(fn func(context.Context, extension.Pr m.projectEvidenceReader = fn } +func (m *Manager) SetProjectContextSourceRechecker(fn func(context.Context, extension.ProjectContextSourceRecheckRequest) (extension.ProjectContextSourceRecheckResult, error)) { + m.projectSourceRechecker = fn +} + func (c *runCtx) ProjectContextEnabled() bool { return c.run.ProjectID != "" && c.mgr.projectContextResolver != nil } @@ -44,12 +49,114 @@ func (c *runCtx) ValidateProjectCrossGuard(ctx context.Context, current extensio if err != nil { return err } + terminal := map[string]string{} for _, m := range prior { + if m.Delivery == "responded" || m.Delivery == "unknown" { + terminal[m.AttemptID] = m.Delivery + } + } + for _, m := range prior { + if m.Delivery == "attempted" { + if terminal[m.AttemptID] != "responded" { + return errors.New("prior project model attempt has no trusted response") + } + continue + } + if m.Delivery != "responded" { + return errors.New("prior project model delivery is unknown") + } 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") } + if err := c.recheckManifest(ctx, m, ""); err != nil { + return err + } + } + return nil +} + +// recheckManifest reauthorizes the exact source identities that were actually +// present in a prior model request. It never persists or reconstructs a Goal. +func (c *runCtx) recheckManifest(ctx context.Context, m extension.ProjectContextManifest, stepID string) error { + if m.Delivery != "responded" || c.mgr.projectSourceRechecker == nil { + return errors.New("project source recheck unavailable") + } + if m.BudgetChars <= 0 || len(m.Included)+len(m.OfferedReferences)+len(m.ReadEvidence) == 0 { + return errors.New("prior project model input has no stable source") + } + for _, entry := range m.Included { + if entry.SourceID == "" || entry.Revision == "" || entry.SourceHash == "" || entry.InputSHA256 == "" { + return errors.New("prior project model input has incomplete source identity") + } + } + for _, ref := range m.OfferedReferences { + if ref.ID == "" || ref.SourceID == "" || ref.Revision == "" || ref.Hash == "" { + return errors.New("prior project evidence reference is incomplete") + } + } + scope, err := c.trustedProjectContextRequest(ctx, m.BudgetChars) + if err != nil { + return err + } + scope.StepID = core.ID(stepID) + result, err := c.mgr.projectSourceRechecker(ctx, extension.ProjectContextSourceRecheckRequest{ + Scope: scope, Included: append([]extension.ProjectContextManifestEntry(nil), m.Included...), + OfferedReferences: append([]extension.ProjectEvidenceRef(nil), m.OfferedReferences...), + ReadEvidence: append([]extension.ProjectContextManifestEntry(nil), m.ReadEvidence...), + }) + if err != nil { + return err + } + current := result.Bundle + expectedRefs := map[string]extension.ProjectEvidenceRef{} + for _, ref := range m.OfferedReferences { + expectedRefs[ref.ID] = ref + } + for _, used := range m.ReadEvidence { + ref := extension.ProjectEvidenceRef{ID: used.ReferenceID, SourceID: used.SourceID, Revision: used.Revision, Hash: used.SourceHash} + if ref.ID == "" || ref.SourceID == "" || ref.Revision == "" || ref.Hash == "" { + return errors.New("read project evidence lacks stable reference") + } + if existing, ok := expectedRefs[ref.ID]; ok && existing != ref { + return errors.New("project evidence reference identity changed") + } + expectedRefs[ref.ID] = ref + } + if len(current.Items) != len(m.Included) || len(current.Evidence) != len(expectedRefs) || !slices.Equal(current.Missing, m.Missing) { + return errors.New("project source recheck returned an incomplete source set") + } + seenRefs := map[string]bool{} + for _, ref := range current.Evidence { + if ref != expectedRefs[ref.ID] || seenRefs[ref.ID] { + return errors.New("project source recheck returned an unrelated evidence reference") + } + seenRefs[ref.ID] = true + } + if err := projectcontext.ValidateCurrent(m, current); err != nil { + return err + } + for _, used := range m.ReadEvidence { + if used.ReferenceID == "" || used.InputSHA256 == "" || c.mgr.projectEvidenceReader == nil { + return errors.New("project evidence recheck 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 + } + readScope := scope + readScope.BudgetChars = 8000 + now, err := c.mgr.projectEvidenceReader(ctx, extension.ProjectEvidenceRequest{Scope: readScope, 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 evidence permission or content changed") + } } return nil } @@ -135,41 +242,50 @@ func (c *runCtx) ProjectContext(ctx context.Context, goal string, budget int) (e if c.run.ProjectID == "" || c.mgr.projectContextResolver == nil { return extension.ProjectContextBundle{}, false, nil } - if c.mgr.projectWorkAuthorization == nil { - return extension.ProjectContextBundle{}, true, errors.New("project work authorization missing") - } - run, err := c.mgr.st.GetRunByID(string(c.run.ID)) + req, err := c.trustedProjectContextRequest(ctx, budget) if err != nil { return extension.ProjectContextBundle{}, true, err } - if run.Owner == "" || run.Owner != c.run.Owner || run.ProjectID != c.run.ProjectID || run.ConversationID != c.run.ConversationID { - return extension.ProjectContextBundle{}, true, errors.New("persisted run scope mismatch") - } - conv, err := c.mgr.st.GetConversation(string(run.Owner), string(run.ConversationID)) - if err != nil { - return extension.ProjectContextBundle{}, true, err - } - if conv.ProjectID != run.ProjectID { - return extension.ProjectContextBundle{}, true, errors.New("persisted conversation scope mismatch") - } - if err := c.mgr.authorizeProjectWork(ctx, string(run.Owner), run.ProjectID); err != nil { - return extension.ProjectContextBundle{}, true, err - } - req := extension.ProjectContextRequest{ - Actor: string(run.Owner), ProjectID: run.ProjectID, ConversationID: run.ConversationID, - RunID: run.ID, Goal: goal, BudgetChars: budget, - } + req.Goal = goal bundle, err := c.mgr.projectContextResolver(ctx, req) if err != nil { return extension.ProjectContextBundle{}, true, err } - if bundle.ProjectID != run.ProjectID || bundle.Status != extension.ProjectContextReady { + if bundle.ProjectID != req.ProjectID || bundle.Status != extension.ProjectContextReady { return extension.ProjectContextBundle{}, true, errors.New("project context unavailable for trusted run") } c.setContextRequest(req) return bundle, true, nil } +func (c *runCtx) trustedProjectContextRequest(ctx context.Context, budget int) (extension.ProjectContextRequest, error) { + if c.mgr.projectWorkAuthorization == nil { + return extension.ProjectContextRequest{}, errors.New("project work authorization missing") + } + run, err := c.mgr.st.GetRunByID(string(c.run.ID)) + if err != nil { + return extension.ProjectContextRequest{}, err + } + if run.Owner == "" || run.Owner != c.run.Owner || run.ProjectID != c.run.ProjectID || run.ConversationID != c.run.ConversationID { + return extension.ProjectContextRequest{}, errors.New("persisted run scope mismatch") + } + conv, err := c.mgr.st.GetConversation(string(run.Owner), string(run.ConversationID)) + if err != nil { + return extension.ProjectContextRequest{}, err + } + if conv.ProjectID != run.ProjectID { + return extension.ProjectContextRequest{}, errors.New("persisted conversation scope mismatch") + } + if err := c.mgr.authorizeProjectWork(ctx, string(run.Owner), run.ProjectID); err != nil { + return extension.ProjectContextRequest{}, err + } + req := extension.ProjectContextRequest{ + Actor: string(run.Owner), ProjectID: run.ProjectID, ConversationID: run.ConversationID, + RunID: run.ID, BudgetChars: budget, + } + return req, nil +} + // SaveContextManifest persists a receipt for the exact request prepared by the // runtime. It does not store model-visible text or candidate-only references. func (c *runCtx) SaveContextManifest(ctx context.Context, m extension.ProjectContextManifest) error { @@ -258,7 +374,7 @@ 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}) + store.ContextManifestScope{Bound: true}) } return c.mgr.st.AppendContextManifest(string(run.Owner), string(run.ID), m) } @@ -309,8 +425,8 @@ func (c *runCtx) VerifyProjectStepOutput(ctx context.Context, stepID, stepAttemp } matchedBound := false for _, source := range receipts { - receipt, selector := source.Manifest, source.Scope - if !selector.Bound || selector.TargetID != "" || receipt.ProjectID != c.run.ProjectID || + receipt := source.Manifest + if !source.Scope.Bound || 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") @@ -318,35 +434,9 @@ func (c *runCtx) VerifyProjectStepOutput(ctx context.Context, stepID, stepAttemp 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 { + if err := c.recheckManifest(ctx, receipt, stepID); 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 := 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") - } - } } if !matchedBound { return errors.New("project workflow output binding lost its responded model attempt") diff --git a/internal/runmgr/project_context_test.go b/internal/runmgr/project_context_test.go index 290802f..3e7b938 100644 --- a/internal/runmgr/project_context_test.go +++ b/internal/runmgr/project_context_test.go @@ -2,6 +2,7 @@ package runmgr import ( "context" + "crypto/sha256" "fmt" "os" "strings" @@ -19,6 +20,69 @@ import ( core "agent-system-core" ) +func TestProjectSourceRecheckVerifiesIncludedRefsAndEvidenceBody(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 }) + item := extension.ProjectContextItem{SourceID: "source-a", Revision: "r1", Hash: "ha", Text: "AUTHORIZED_A", Required: true} + ref := extension.ProjectEvidenceRef{ID: "ref-a", SourceID: "doc-a", Revision: "r1", Hash: "hd"} + bundle := extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: run.ProjectID, Version: "v1", Watermark: "w1", AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{item}, Evidence: []extension.ProjectEvidenceRef{ref}} + _, receipt, err := projectcontext.Assemble(bundle, 900) + if err != nil { + t.Fatal(err) + } + receipt.Delivery = "responded" + receipt.ReadEvidence = []extension.ProjectContextManifestEntry{{ReferenceID: ref.ID, SourceID: ref.SourceID, Revision: ref.Revision, SourceHash: ref.Hash, InputSHA256: fmt.Sprintf("%x", sha256.Sum256([]byte("DETAIL_A")))}} + rc := &runCtx{mgr: mgr, run: run, ctx: context.Background()} + current := bundle + readBody := "DETAIL_A" + offeredCount := 1 + mgr.SetProjectContextSourceRechecker(func(_ context.Context, req extension.ProjectContextSourceRecheckRequest) (extension.ProjectContextSourceRecheckResult, error) { + if req.Scope.Actor != string(run.Owner) || req.Scope.ProjectID != run.ProjectID || req.Scope.RunID != run.ID || req.Scope.ConversationID != run.ConversationID || req.Scope.StepID != "draft" || req.Scope.Goal != "" || len(req.Included) != 1 || len(req.OfferedReferences) != offeredCount || len(req.ReadEvidence) != 1 { + return extension.ProjectContextSourceRecheckResult{}, fmt.Errorf("source recheck lost trusted run scope or actual input refs") + } + return extension.ProjectContextSourceRecheckResult{Bundle: current}, nil + }) + mgr.SetProjectEvidenceReader(func(_ context.Context, req extension.ProjectEvidenceRequest) (extension.ProjectEvidence, error) { + if req.Scope.Actor != string(run.Owner) || req.Scope.RunID != run.ID || req.Scope.StepID != "draft" || req.Scope.Goal != "" || req.Reference != ref { + return extension.ProjectEvidence{}, fmt.Errorf("evidence read lost current trusted scope") + } + return extension.ProjectEvidence{Reference: req.Reference, Text: readBody}, nil + }) + if err := rc.recheckManifest(context.Background(), receipt, "draft"); err != nil { + t.Fatalf("same authorized sources rejected: %v", err) + } + // A detailed body may be in the final request even when its reference was + // not offered in that request's short context. Recheck that exact read too. + receipt.OfferedReferences = nil + offeredCount = 0 + if err := rc.recheckManifest(context.Background(), receipt, "draft"); err != nil { + t.Fatalf("read evidence not offered as a short ref was rejected: %v", err) + } + receipt.OfferedReferences = []extension.ProjectEvidenceRef{ref} + offeredCount = 1 + current.Items = []extension.ProjectContextItem{{SourceID: item.SourceID, Revision: item.Revision, Hash: item.Hash, Text: "CHANGED", Required: true}} + if err := rc.recheckManifest(context.Background(), receipt, "draft"); err == nil { + t.Fatal("provider echoed old source hash but changed model-visible text") + } + current = bundle + current.Evidence = nil + if err := rc.recheckManifest(context.Background(), receipt, "draft"); err == nil { + t.Fatal("withdrawn offered reference accepted") + } + current = bundle + readBody = "REVOKED_DETAIL" + if err := rc.recheckManifest(context.Background(), receipt, "draft"); err == nil { + t.Fatal("changed ReadEvidence body accepted") + } + mgr.SetProjectContextSourceRechecker(nil) + if err := rc.recheckManifest(context.Background(), receipt, "draft"); err == nil { + t.Fatal("missing rechecker accepted old machine context") + } +} + func TestProjectContextUsesPersistedRunAndRefreshesAuthorization(t *testing.T) { st := openRecoveryStore(t) m := newRecoveryManager(t, st) @@ -90,6 +154,10 @@ func TestProjectJudgeChecksPersistedSourceEpochAcrossGuards(t *testing.T) { 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 }) + m.SetProjectContextSourceRechecker(func(_ context.Context, req extension.ProjectContextSourceRecheckRequest) (extension.ProjectContextSourceRecheckResult, error) { + return extension.ProjectContextSourceRecheckResult{Bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.Scope.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) @@ -182,6 +250,14 @@ func TestProjectJudgeRetainsGoalSensitiveSourceScope(t *testing.T) { } return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID, AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Watermark: "w1", Items: []extension.ProjectContextItem{item}}, nil }) + m.SetProjectContextSourceRechecker(func(_ context.Context, req extension.ProjectContextSourceRecheckRequest) (extension.ProjectContextSourceRecheckResult, error) { + item := extension.ProjectContextItem{SourceID: "source-b", Revision: "r1", Hash: "hb", Text: "PUBLIC_B", Required: true} + if len(req.Included) == 1 && req.Included[0].SourceID == "source-a" && available { + item = extension.ProjectContextItem{SourceID: "source-a", Revision: "r1", Hash: "ha", Text: "AUTHORIZED_A", Required: true} + } + return extension.ProjectContextSourceRecheckResult{Bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.Scope.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) @@ -238,10 +314,12 @@ func TestProjectJudgeRetainsGoalSensitiveSourceScope(t *testing.T) { type goalSensitiveStepBackend struct { stream agentllm.StreamFunc sent string + calls int } func (*goalSensitiveStepBackend) Name() string { return "goal-sensitive-test" } func (b *goalSensitiveStepBackend) Execute(ctx context.Context, req squad.ExecRequest) (*squad.ExecResult, error) { + b.calls++ 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 @@ -254,6 +332,52 @@ func (b *goalSensitiveStepBackend) Execute(ctx context.Context, req squad.ExecRe return &squad.ExecResult{Success: true, Patches: map[string][]byte{"output.md": []byte("AUTHORIZED_A report\n")}}, nil } +func TestCompiledWorkflowSecondPromptRejectsRevokedPriorOutput(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 + item := extension.ProjectContextItem{SourceID: "source-a", Revision: "r1", Hash: "ha", Text: "AUTHORIZED_A", Required: true} + mgr.SetProjectContextResolver(func(_ context.Context, req extension.ProjectContextRequest) (extension.ProjectContextBundle, error) { + items := []extension.ProjectContextItem{item} + if !available { + items = []extension.ProjectContextItem{{SourceID: "source-b", Revision: "r1", Hash: "hb", Text: "PUBLIC_B", Required: true}} + } + return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID, AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Watermark: "w1", Items: items}, nil + }) + rechecks := 0 + mgr.SetProjectContextSourceRechecker(func(_ context.Context, req extension.ProjectContextSourceRecheckRequest) (extension.ProjectContextSourceRecheckResult, error) { + rechecks++ + result := extension.ProjectContextSourceRecheckResult{Bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.Scope.ProjectID, AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Watermark: "w1", Items: []extension.ProjectContextItem{item}}} + if rechecks == 1 { + available = false + } else { + result.Bundle.Items = nil + } + return result, nil + }) + rc := &runCtx{mgr: mgr, run: run, ctx: context.Background()} + base := func(_ context.Context, _ string, _ []agentllm.Message, _ []agentllm.ToolDecl, _ agentllm.StreamOptions) <-chan agentllm.Event { + ch := make(chan agentllm.Event, 1) + ch <- agentllm.Event{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: "done"}} + close(ch) + return ch + } + backend := &goalSensitiveStepBackend{stream: runtime.GuardProjectStream(rc, "", base)} + backends := squad.NewBackendResolver() + backends.RegisterInstance(backend.Name(), backend) + def := core.WorkflowDef{ID: "two-prompts", Title: "Two prompts", Steps: []core.WorkflowStep{ + {ID: "draft", Title: "Draft", Kind: "prompt", Agent: "writer", Prompt: "draft report"}, + {ID: "rewrite", Title: "Rewrite", Kind: "prompt", Agent: "writer", Prompt: "revise {{steps.draft.output}}"}, + }} + _, err := wfcompile.RunCompiled(rc, wfcompile.Deps{DataDir: t.TempDir(), Backends: backends, DefaultBackend: backend.Name()}, workflow.Launch{Def: def}, workflow.Vars{Input: "draft"}, wfcompile.RunOptions{}) + if err == nil || backend.calls != 1 || rechecks < 2 { + t.Fatalf("second prompt consumed revoked first output: err=%v calls=%d rechecks=%d", err, backend.calls, rechecks) + } +} + func TestCompiledWorkflowJudgeAcceptsGoalSensitiveProviderAndRejectsWithdrawal(t *testing.T) { st := openRecoveryStore(t) mgr := newRecoveryManager(t, st) @@ -270,6 +394,14 @@ func TestCompiledWorkflowJudgeAcceptsGoalSensitiveProviderAndRejectsWithdrawal(t } return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID, AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Watermark: "w1", Items: []extension.ProjectContextItem{item}}, nil }) + mgr.SetProjectContextSourceRechecker(func(_ context.Context, req extension.ProjectContextSourceRecheckRequest) (extension.ProjectContextSourceRecheckResult, error) { + item := extension.ProjectContextItem{SourceID: "source-b", Revision: "r1", Hash: "hb", Text: "PUBLIC_B", Required: true} + if len(req.Included) == 1 && req.Included[0].SourceID == "source-a" && available { + item = extension.ProjectContextItem{SourceID: "source-a", Revision: "r1", Hash: "ha", Text: "AUTHORIZED_A", Required: true} + } + return extension.ProjectContextSourceRecheckResult{Bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.Scope.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) @@ -298,6 +430,26 @@ func TestCompiledWorkflowJudgeAcceptsGoalSensitiveProviderAndRejectsWithdrawal(t 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) } + rows, err := st.DB().Query(`SELECT manifest_json FROM context_manifests WHERE owner=$1 AND target_id=$2`, string(run.Owner), string(run.ID)) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var raw string + if err := rows.Scan(&raw); err != nil { + rows.Close() + t.Fatal(err) + } + if strings.Contains(raw, "AUTHORIZED_A report") || strings.Contains(raw, "draft report using") { + rows.Close() + t.Fatal("workflow model prompt or output body entered trusted receipt JSON") + } + } + if err := rows.Err(); err != nil { + rows.Close() + t.Fatal(err) + } + rows.Close() bound, err := st.WorkflowStepSource(string(run.Owner), string(run.ID), "draft") if err != nil { t.Fatal(err) @@ -337,7 +489,7 @@ func TestContextManifestRejectsDocumentDutyRevocationBeforeSend(t *testing.T) { epoch := "duty-1" m.SetProjectContextResolver(func(_ context.Context, req extension.ProjectContextRequest) (extension.ProjectContextBundle, error) { return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID, Version: "v1", AuthorizationEpoch: epoch, - Items: []extension.ProjectContextItem{{SourceID: "restricted-doc", Text: "private text", Required: true}}}, nil + Items: []extension.ProjectContextItem{{SourceID: "restricted-doc", Revision: "r1", Hash: "h1", Text: "private text", Required: true}}}, nil }) rc := &runCtx{mgr: m, run: run, ctx: context.Background()} bundle, active, err := rc.ProjectContext(context.Background(), "work", 900) @@ -410,7 +562,7 @@ func TestConcurrentProjectEvidenceReadsKeepTrustedScope(t *testing.T) { conv := recoveryConversation(t, st) run := recoveryRun(t, st, conv, core.RunPending) m.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return true, nil }) - ref := extension.ProjectEvidenceRef{ID: "r1", SourceID: "doc", Revision: "v1"} + ref := extension.ProjectEvidenceRef{ID: "r1", SourceID: "doc", Revision: "v1", Hash: "h1"} m.SetProjectContextResolver(func(_ context.Context, req extension.ProjectContextRequest) (extension.ProjectContextBundle, error) { return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID, Evidence: []extension.ProjectEvidenceRef{ref}}, nil }) diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index 1ed8485..fee313a 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -64,6 +64,7 @@ type Manager struct { projectWorkAuthorization func(context.Context, string, core.ID) (bool, error) projectContextResolver func(context.Context, extension.ProjectContextRequest) (extension.ProjectContextBundle, error) projectEvidenceReader func(context.Context, extension.ProjectEvidenceRequest) (extension.ProjectEvidence, error) + projectSourceRechecker func(context.Context, extension.ProjectContextSourceRecheckRequest) (extension.ProjectContextSourceRecheckResult, error) // commands EXT-14:命令即工具的投影与分发源(nil = 未接线,命令不进工具清单)。 commands CommandSource indexer core.DocumentIndexer // 可选索引入口(nil = store 整文档单块) diff --git a/internal/runtime/project_context_test.go b/internal/runtime/project_context_test.go index 06299fe..0edb559 100644 --- a/internal/runtime/project_context_test.go +++ b/internal/runtime/project_context_test.go @@ -65,7 +65,7 @@ func TestProjectContextRefreshesPerModelRequestWithoutPersistingOldSummary(t *te return base(ctx, model, msgs, tools, opts) }} c := &projectRunCtx{fakeRunCtx: &fakeRunCtx{ctx: context.Background(), sandbox: t.TempDir(), projectID: "project-1", prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}}}} - c.bundle = extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "project-1", Version: "v1", Items: []extension.ProjectContextItem{{SourceID: "s1", Text: "summary-v1", Required: true}}} + c.bundle = extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "project-1", Version: "v1", Items: []extension.ProjectContextItem{{SourceID: "s1", Revision: "r1", Hash: "h1", Text: "summary-v1", Required: true}}} if err := runner.Execute(c); err != nil { t.Fatal(err) } @@ -110,7 +110,7 @@ func TestProjectRevocationScrubsOldMachineContextBeforeSecondTurn(t *testing.T) base := scriptStream(turns, nil) var requests [][]agentllm.Message c := &projectRunCtx{fakeRunCtx: &fakeRunCtx{ctx: context.Background(), projectID: "p1", prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}}, memEntries: []memEntry{{typ: "note", content: "STALE_MEMORY"}}, execOutput: old}, - bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", Version: "v1", AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{{SourceID: "doc", Text: old, Required: true}}}, + bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", Version: "v1", AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{{SourceID: "doc", Revision: "r1", Hash: "h1", Text: old, Required: true}}}, budgetSettings: core.EffectiveSettings{"models": map[string]any{"chat": map[string]any{"baseUrl": "http://test"}}, "context": map[string]any{"userInstructions": "STALE_INSTRUCTIONS"}, "runtime": map[string]any{"verify": false, "composeFinal": false}}, sessionNote: "STALE_SESSION_NOTE", injections: []ContextInjection{{Content: "STALE_INJECTION", Constant: true}}, surfaces: []core.SurfaceSnapshot{{SemanticText: "STALE_SURFACE"}}, @@ -118,7 +118,7 @@ func TestProjectRevocationScrubsOldMachineContextBeforeSecondTurn(t *testing.T) runner := &LLMRunner{Stream: func(ctx context.Context, model string, msgs []agentllm.Message, tools []agentllm.ToolDecl, opts agentllm.StreamOptions) <-chan agentllm.Event { requests = append(requests, append([]agentllm.Message(nil), msgs...)) if len(requests) == 1 { - c.bundle = extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", Version: "v2", AuthorizationEpoch: "duty-2", PublicationEpoch: "pub-2", Items: []extension.ProjectContextItem{{SourceID: "public", Text: "CURRENT_PUBLIC_FACT", Required: true}}} + c.bundle = extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", Version: "v2", AuthorizationEpoch: "duty-2", PublicationEpoch: "pub-2", Items: []extension.ProjectContextItem{{SourceID: "public", Revision: "r1", Hash: "h2", Text: "CURRENT_PUBLIC_FACT", Required: true}}} } return base(ctx, model, msgs, tools, opts) }} @@ -160,7 +160,7 @@ func TestProjectSameScopeKeepsCurrentRunToolContext(t *testing.T) { return base(ctx, model, msgs, tools, opts) }} c := &projectRunCtx{fakeRunCtx: &fakeRunCtx{ctx: context.Background(), projectID: "p1", prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}}, execOutput: "AUTHORIZED_TOOL_RESULT"}, - bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{{SourceID: "fact", Text: "CURRENT_FACT", Required: true}}}} + bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{{SourceID: "fact", Revision: "r1", Hash: "h1", Text: "CURRENT_FACT", Required: true}}}} if err := runner.Execute(c); err != nil { t.Fatal(err) } @@ -193,7 +193,7 @@ func TestSubagentModelRequestUsesProjectContextAndManifest(t *testing.T) { return base(ctx, model, msgs, tools, opts) }} c := &projectRunCtx{fakeRunCtx: &fakeRunCtx{ctx: context.Background(), projectID: "project-1", prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model", Lightweight: "test-lite"}}}, - bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "project-1", Items: []extension.ProjectContextItem{{SourceID: "s1", Text: "current project fact", Required: true}}}} + bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "project-1", Items: []extension.ProjectContextItem{{SourceID: "s1", Revision: "r1", Hash: "h1", Text: "current project fact", Required: true}}}} if err := sub.Execute(c); err != nil { t.Fatal(err) } @@ -214,7 +214,7 @@ func TestSubagentSameScopeKeepsNewToolResultAcrossTurns(t *testing.T) { return base(ctx, model, msgs, tools, opts) }} c := &projectRunCtx{fakeRunCtx: &fakeRunCtx{ctx: context.Background(), projectID: "p1", prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model", Lightweight: "test-lite"}}, execOutput: "NEW_AUTHORIZED_TOOL"}, - bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{{SourceID: "fact", Text: "CURRENT_FACT", Required: true}}}, + bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{{SourceID: "fact", Revision: "r1", Hash: "h1", Text: "CURRENT_FACT", Required: true}}}, toolList: []core.ToolDesc{{Name: "knowledge_search", Description: "Search", Parameters: json.RawMessage(`{"type":"object"}`)}}, } if err := sub.Execute(c); err != nil { @@ -248,7 +248,7 @@ func TestProjectStreamBlocksRequestExceedingFinalModelBudget(t *testing.T) { called = true return scriptStream(textTurns("bad"), nil)(ctx, model, msgs, tools, opts) } - c := &projectRunCtx{fakeRunCtx: &fakeRunCtx{projectID: "p1"}, bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", Items: []extension.ProjectContextItem{{SourceID: "large", Text: strings.Repeat("x", 11000), Required: true}}}, budgetSettings: core.EffectiveSettings{"models": map[string]any{"chat": map[string]any{"contextWindow": float64(4096), "outputReserve": float64(512), "contextBuffer": float64(512)}}}} + c := &projectRunCtx{fakeRunCtx: &fakeRunCtx{projectID: "p1"}, bundle: extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "p1", Items: []extension.ProjectContextItem{{SourceID: "large", Revision: "r1", Hash: "h1", Text: strings.Repeat("x", 11000), Required: true}}}, budgetSettings: core.EffectiveSettings{"models": map[string]any{"chat": map[string]any{"contextWindow": float64(4096), "outputReserve": float64(512), "contextBuffer": float64(512)}}}} for ev := range GuardProjectStream(c, "goal", base)(context.Background(), "test", []agentllm.Message{{Role: agentllm.RoleUser, Content: "goal"}}, nil, agentllm.StreamOptions{}) { if ev.Kind != agentllm.KindError { t.Fatalf("oversized request did not fail: %+v", ev) @@ -335,7 +335,7 @@ func (c *projectSquadCtx) Run() core.Run { return core.Run{ID: "run-squad", Owner: "alice", ProjectID: "project-1"} } func (c *projectSquadCtx) 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: "squad-source", Text: "squad project fact", Required: true}}}, true, nil + return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "project-1", AuthorizationEpoch: "duty-1", PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{{SourceID: "squad-source", Revision: "r1", Hash: "h1", Text: "squad project fact", Required: true}}}, true, nil } func (c *projectSquadCtx) SaveContextManifest(_ context.Context, m extension.ProjectContextManifest) error { c.manifests = append(c.manifests, m) diff --git a/internal/runtime/workflow_compiled_test.go b/internal/runtime/workflow_compiled_test.go index 51b1301..1f7148f 100644 --- a/internal/runtime/workflow_compiled_test.go +++ b/internal/runtime/workflow_compiled_test.go @@ -308,7 +308,7 @@ func (c *projectJudgeCtx) ProjectContext(context.Context, string, int) (extensio 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 + return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "project-1", AuthorizationEpoch: epoch, PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{{SourceID: "judge-source", Revision: "r1", Hash: "h1", Text: "current project rule", Required: true}}}, true, nil } func (c *projectJudgeCtx) SaveContextManifest(_ context.Context, m extension.ProjectContextManifest) error { c.manifests = append(c.manifests, m) diff --git a/internal/store/context_manifest.go b/internal/store/context_manifest.go index 779f7b2..48e52a9 100644 --- a/internal/store/context_manifest.go +++ b/internal/store/context_manifest.go @@ -11,12 +11,10 @@ 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. +// ContextManifestScope records only that the request came from a trusted model +// guard. Raw goals and model output are never persisted in this receipt. type ContextManifestScope struct { - Goal, TargetID string - Bound bool + Bound bool } // AppendContextManifest writes to the trusted receipt table, never to the @@ -50,17 +48,16 @@ func (s *Store) appendContextManifest(owner, targetID string, m extension.Projec } defer tx.Rollback() if m.Delivery != "attempted" { - 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) + err := tx.QueryRow(s.q(`SELECT 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(&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 { + if (attemptedBound != 0) != scope.Bound { return errors.New("context manifest terminal resolve scope changed") } var terminal int @@ -76,8 +73,8 @@ func (s *Store) appendContextManifest(owner, targetID string, m extension.Projec 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) + _, err = tx.Exec(s.q(`INSERT INTO context_manifests(id,attempt_id,owner,target_id,project_id,input_sha256,delivery,manifest_json,created_at,scope_bound) VALUES(?,?,?,?,?,?,?,?,?,?)`), + NewID("ctxm"), m.AttemptID, owner, targetID, m.ProjectID, m.InputSHA256, m.Delivery, string(raw), nowISO(), 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 c4411b8..2259897 100644 --- a/internal/store/context_manifest_test.go +++ b/internal/store/context_manifest_test.go @@ -1,7 +1,10 @@ package store import ( + "bytes" + "os" "path/filepath" + "strings" "testing" "agent-system-backend/pkg/extension" @@ -37,6 +40,132 @@ func TestContextManifestTerminalRequiresMatchingAttempt(t *testing.T) { } } +func TestContextManifestDoesNotPersistWorkflowOutputInResolveScope(t *testing.T) { + st, err := OpenSQLite(filepath.Join(t.TempDir(), "scope-privacy.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + const restricted = "OLD_RESTRICTED_BODY" + m := extension.ProjectContextManifest{ProjectID: "project-1", AttemptID: "judge-attempt", InputSHA256: "input-hash", Delivery: "attempted"} + scope := ContextManifestScope{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, scope); err != nil { + t.Fatal(err) + } + rows, err := st.db.Query(st.q(`SELECT delivery,manifest_json FROM context_manifests WHERE attempt_id=? ORDER BY created_at,id`), m.AttemptID) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + seen := 0 + for rows.Next() { + var delivery, manifest string + if err := rows.Scan(&delivery, &manifest); err != nil { + t.Fatal(err) + } + seen++ + if strings.Contains(manifest, restricted) { + t.Errorf("%s trusted receipt persisted workflow output body", delivery) + } + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + if seen != 2 { + t.Fatalf("expected attempted and responded receipts, got %d", seen) + } + var rawColumns int + if err := st.db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('context_manifests') WHERE name IN ('resolve_goal','resolve_target_id')`).Scan(&rawColumns); err != nil || rawColumns != 0 { + t.Fatalf("fresh receipt schema retains raw scope columns: columns=%d err=%v", rawColumns, err) + } +} + +func TestCandidateV34RawScopeColumnsAreScrubbedOnUpgrade(t *testing.T) { + path := filepath.Join(t.TempDir(), "candidate-v34.db") + st, err := OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + m := extension.ProjectContextManifest{ProjectID: "p1", AttemptID: "old-model", InputSHA256: "input", Delivery: "attempted"} + if err := st.AppendContextManifestWithScope("alice", "run-1", m, ContextManifestScope{Bound: true}); err != nil { + t.Fatal(err) + } + for _, query := range []string{ + `DELETE FROM schema_migrations WHERE version=35`, + `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 ''`, + `UPDATE context_manifests SET resolve_goal='OLD_RESTRICTED_BODY',resolve_target_id='doc-private'`, + } { + if _, err := st.db.Exec(query); err != nil { + t.Fatal(err) + } + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + upgraded, err := OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + defer upgraded.Close() + var columns int + if err := upgraded.db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('context_manifests') WHERE name IN ('resolve_goal','resolve_target_id')`).Scan(&columns); err != nil || columns != 0 { + t.Fatalf("candidate raw scope columns survived v35: columns=%d err=%v", columns, err) + } + var count int + if err := upgraded.db.QueryRow(`SELECT COUNT(*) FROM context_manifests WHERE attempt_id='old-model'`).Scan(&count); err != nil || count != 1 { + t.Fatalf("trusted manifest disappeared during scrub: count=%d err=%v", count, err) + } + for _, suffix := range []string{"", "-wal"} { + data, err := os.ReadFile(path + suffix) + if err == nil && bytes.Contains(data, []byte("OLD_RESTRICTED_BODY")) { + t.Fatalf("candidate raw scope bytes remain in SQLite %s after upgrade", suffix) + } + } +} + +func TestCandidateV34PostgresRawScopeScrubAndFreshSchema(t *testing.T) { + st := openTestStore(t) + if st.Dialect() != "postgres" { + t.Skip("isolated PostgreSQL DSN not configured") + } + checkColumns := func(want int) { + t.Helper() + var count int + if err := st.db.QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='context_manifests' AND column_name IN ('resolve_goal','resolve_target_id')`).Scan(&count); err != nil || count != want { + t.Fatalf("raw scope columns=%d want=%d err=%v", count, want, err) + } + } + checkColumns(0) + m := extension.ProjectContextManifest{ProjectID: "p1", AttemptID: "candidate-model", InputSHA256: "input", Delivery: "attempted"} + if err := st.AppendContextManifestWithScope("alice", "run-1", m, ContextManifestScope{Bound: true}); err != nil { + t.Fatal(err) + } + for _, query := range []string{ + `DELETE FROM schema_migrations WHERE version=35`, + `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 ''`, + `UPDATE context_manifests SET resolve_goal='OLD_RESTRICTED_BODY',resolve_target_id='doc-private'`, + } { + if _, err := st.db.Exec(query); err != nil { + t.Fatal(err) + } + } + checkColumns(2) + if err := st.migrate(); err != nil { + t.Fatal(err) + } + checkColumns(0) + var count int + if err := st.db.QueryRow(`SELECT COUNT(*) FROM context_manifests WHERE attempt_id='candidate-model'`).Scan(&count); err != nil || count != 1 { + t.Fatalf("trusted manifest lost during PG scrub: count=%d err=%v", count, err) + } +} + func TestContextManifestTerminalCannotChangeTrustedResolveScope(t *testing.T) { st, err := OpenSQLite(filepath.Join(t.TempDir(), "scope.db")) if err != nil { @@ -44,14 +173,11 @@ func TestContextManifestTerminalCannotChangeTrustedResolveScope(t *testing.T) { } 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} + scope := ContextManifestScope{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") } diff --git a/internal/store/migrate.go b/internal/store/migrate.go index bf6fab9..b9bb9af 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -3,6 +3,7 @@ package store import ( "context" "database/sql" + "errors" "fmt" "strings" "time" @@ -871,14 +872,18 @@ func init() { }, 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`}, + "sqlite": {`ALTER TABLE context_manifests ADD COLUMN scope_bound INTEGER NOT NULL DEFAULT 0`}, + "postgres": {`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`}, + "sqlite": {`ALTER TABLE context_manifests DROP COLUMN scope_bound`}, + "postgres": {`ALTER TABLE context_manifests DROP COLUMN scope_bound`}, }, }, + migration{version: 35, name: "scrub_candidate_context_scope_text", + up: map[string][]string{"sqlite": {`SELECT 1`}, "postgres": {`SELECT 1`}}, + down: map[string][]string{"sqlite": {`SELECT 1`}, "postgres": {`SELECT 1`}}, + }, ) } @@ -971,10 +976,35 @@ func (s *Store) migrate() error { if s.dialect.name() == "postgres" { return s.migratePG() } + var candidateRawColumns int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('context_manifests') WHERE name IN ('resolve_goal','resolve_target_id')`).Scan(&candidateRawColumns); err != nil { + return err + } // SQLite:_txlock=immediate(OpenSQLite DSN)保证迁移事务在 BEGIN 即取写锁, // 跨进程迁移者按 busy_timeout 排队;后续版本读取发生在同一事务内, // 前一个迁移者提交后本进程看到的已是完整账本(无部分版本可见)。 - return s.migratePass() + if err := s.migratePass(); err != nil { + return err + } + if candidateRawColumns > 0 { + // The prior candidate schema may have stored full prompts. VACUUM is + // outside the DDL transaction; it compacts the isolated SQLite file + // after v35 drops those columns. Refuse to serve if compaction fails. + if _, err := s.db.Exec(`PRAGMA secure_delete=ON`); err != nil { + return fmt.Errorf("candidate context scrub: %w", err) + } + if _, err := s.db.Exec(`VACUUM`); err != nil { + return fmt.Errorf("candidate context scrub: %w", err) + } + var busy, logFrames, checkpointed int + if err := s.db.QueryRow(`PRAGMA wal_checkpoint(TRUNCATE)`).Scan(&busy, &logFrames, &checkpointed); err != nil { + return fmt.Errorf("candidate context scrub checkpoint: %w", err) + } + if busy != 0 { + return errors.New("candidate context scrub checkpoint remains busy") + } + } + return nil } // migratePG 以会话级咨询锁串行化并发迁移者(KTD8):锁在专用连接上获取, @@ -1052,6 +1082,11 @@ func (s *Store) migratePass() error { if applied[m.version] { continue } + if m.version == 35 { + if err := scrubCandidateContextScope(tx, s.dialect.name()); err != nil { + return fmt.Errorf("migrate v35: %w", err) + } + } stmts := m.up[s.dialect.name()] if len(stmts) == 0 { return fmt.Errorf("migrate: v%d %s 缺少 %s 方言", m.version, m.name, s.dialect.name()) @@ -1073,6 +1108,34 @@ func (s *Store) migratePass() error { return tx.Commit() } +// Candidate v34 was never released, but isolated test databases may already +// contain raw Resolve goals. Fresh v34 creates no such columns; v35 removes +// them conditionally from the earlier candidate schema before serving. +func scrubCandidateContextScope(tx *sql.Tx, dialect string) error { + for _, column := range []string{"resolve_goal", "resolve_target_id"} { + var exists int + var err error + if dialect == "sqlite" { + err = tx.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('context_manifests') WHERE name=?`, column).Scan(&exists) + } else { + err = tx.QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='context_manifests' AND column_name=$1`, column).Scan(&exists) + } + if err != nil { + return err + } + if exists == 0 { + continue + } + if _, err := tx.Exec(`UPDATE context_manifests SET ` + column + `=''`); err != nil { + return err + } + if _, err := tx.Exec(`ALTER TABLE context_manifests DROP COLUMN ` + column); err != nil { + return err + } + } + return nil +} + func latestMigrationVersion() int { if len(migrations) == 0 { return 0 diff --git a/internal/store/migrate_compat.go b/internal/store/migrate_compat.go index 7b17ce6..a3addac 100644 --- a/internal/store/migrate_compat.go +++ b/internal/store/migrate_compat.go @@ -39,6 +39,7 @@ var migrationBreaksOldReaders = map[int]bool{ 32: false, // trusted_context_manifests:独立可信回执表,旧读者不受影响 33: false, // workflow_step_sources:新增逐步骤来源绑定表,旧读者不受影响 34: false, // context_manifest_resolve_scope:可信回执增加原Resolve范围,旧读者不受影响 + 35: false, // scrub_candidate_context_scope_text:候选v34原文列清理,旧读者不受影响 } // SchemaTarget 是本二进制会迁移到的 schema 版本。 diff --git a/internal/store/workflow_step_source.go b/internal/store/workflow_step_source.go index e5bbfad..7806c6b 100644 --- a/internal/store/workflow_step_source.go +++ b/internal/store/workflow_step_source.go @@ -40,17 +40,18 @@ 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,resolve_target_id,scope_bound FROM context_manifests WHERE owner=? AND target_id=? AND delivery='responded'`), owner, runID) + rows, err := s.db.Query(s.q(`SELECT created_at,delivery,manifest_json,scope_bound FROM context_manifests WHERE owner=? AND target_id=?`), owner, runID) if err != nil { return WorkflowStepSource{}, err } defer rows.Close() var latest extension.ProjectContextManifest var latestAt time.Time + states := map[string]uint8{} for rows.Next() { - var atRaw, raw, resolveTarget string + var atRaw, delivery, raw string var scopeBound int - if err := rows.Scan(&atRaw, &raw, &resolveTarget, &scopeBound); err != nil { + if err := rows.Scan(&atRaw, &delivery, &raw, &scopeBound); err != nil { return WorkflowStepSource{}, err } at, err := time.Parse(time.RFC3339Nano, atRaw) @@ -67,7 +68,19 @@ func (s *Store) BindWorkflowStepSource(owner, runID, stepID, stepAttemptID, kind if m.WorkflowStepAttemptID != stepAttemptID { continue } - if scopeBound != 1 || resolveTarget != "" { + switch delivery { + case "attempted": + states[m.AttemptID] |= 1 + continue + case "unknown": + states[m.AttemptID] |= 4 + continue + case "responded": + states[m.AttemptID] |= 2 + default: + return WorkflowStepSource{}, errors.New("workflow model receipt has unknown delivery state") + } + if scopeBound != 1 { return WorkflowStepSource{}, errors.New("workflow step model receipt lacks trusted original Resolve scope") } if m.ProjectID != run.ProjectID || m.AuthorizationEpoch == "" || m.PublicationEpoch == "" || m.AttemptID == "" { @@ -83,6 +96,11 @@ func (s *Store) BindWorkflowStepSource(owner, runID, stepID, stepAttemptID, kind if err := rows.Err(); err != nil { return WorkflowStepSource{}, err } + for _, state := range states { + if state != 3 { + return WorkflowStepSource{}, errors.New("workflow step has an unsettled or unknown model attempt") + } + } if latestAt.IsZero() { return WorkflowStepSource{}, errors.New("workflow prompt output has no responded model receipt") } @@ -108,16 +126,17 @@ func (s *Store) RespondedWorkflowStepManifests(owner, runID, stepAttemptID strin 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) + rows, err := s.db.Query(s.q(`SELECT delivery,manifest_json,scope_bound FROM context_manifests WHERE owner=? AND target_id=? ORDER BY created_at,id`), owner, runID) if err != nil { return nil, err } defer rows.Close() var out []ScopedContextManifest + states := map[string]uint8{} for rows.Next() { - var raw, goal, target string + var delivery, raw string var bound int - if err := rows.Scan(&raw, &goal, &target, &bound); err != nil { + if err := rows.Scan(&delivery, &raw, &bound); err != nil { return nil, err } var manifest extension.ProjectContextManifest @@ -127,11 +146,28 @@ func (s *Store) RespondedWorkflowStepManifests(owner, runID, stepAttemptID strin if manifest.WorkflowStepAttemptID != stepAttemptID { continue } - out = append(out, ScopedContextManifest{Manifest: manifest, Scope: ContextManifestScope{Goal: goal, TargetID: target, Bound: bound == 1}}) + switch delivery { + case "attempted": + states[manifest.AttemptID] |= 1 + continue + case "unknown": + states[manifest.AttemptID] |= 4 + continue + case "responded": + states[manifest.AttemptID] |= 2 + default: + return nil, errors.New("workflow model receipt has unknown delivery state") + } + out = append(out, ScopedContextManifest{Manifest: manifest, Scope: ContextManifestScope{Bound: bound == 1}}) } if err := rows.Err(); err != nil { return nil, err } + for _, state := range states { + if state != 3 { + return nil, errors.New("workflow step has an unsettled or unknown model attempt") + } + } if len(out) == 0 { return nil, ErrNotFound } diff --git a/internal/store/workflow_step_source_test.go b/internal/store/workflow_step_source_test.go index f58d9f7..50eed31 100644 --- a/internal/store/workflow_step_source_test.go +++ b/internal/store/workflow_step_source_test.go @@ -30,7 +30,7 @@ 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"} - scope := ContextManifestScope{Goal: "draft report", Bound: true} + scope := ContextManifestScope{Bound: true} if err := st.AppendContextManifestWithScope("alice", string(r.ID), m, scope); err != nil { t.Fatal(err) } @@ -79,6 +79,21 @@ func TestWorkflowStepSourceRequiresRespondedManifestAndExactRunStep(t *testing.T 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") } + unknown := m + unknown.AttemptID, unknown.Delivery = "unknown-model-attempt", "attempted" + if err := st.AppendContextManifestWithScope("alice", string(r.ID), unknown, scope); err != nil { + t.Fatal(err) + } + unknown.Delivery = "unknown" + if err := st.AppendContextManifestWithScope("alice", string(r.ID), unknown, scope); err != nil { + t.Fatal(err) + } + if _, err := st.RespondedWorkflowStepManifests("alice", string(r.ID), "engine-run/draft/1"); err == nil { + t.Fatal("unknown model attempt hidden by earlier responded attempt") + } + if _, err := st.BindWorkflowStepSource("alice", string(r.ID), "draft", "engine-run/draft/1", "prompt", started, []byte("old body")); err == nil { + t.Fatal("unknown model attempt rebound completed step") + } if err := st.UnbindWorkflowStepSource("alice", string(r.ID), "draft"); err != nil { t.Fatal(err) } diff --git a/internal/wfcompile/host.go b/internal/wfcompile/host.go index b0ca13a..12a57da 100644 --- a/internal/wfcompile/host.go +++ b/internal/wfcompile/host.go @@ -98,6 +98,17 @@ type hostRunner struct { } func (r *hostRunner) Run(ctx context.Context, step squad.IntentStep) (*squad.ExecResult, error) { + if r.rc.Run().ProjectID != "" { + guard, ok := r.rc.(ProjectStepProvenance) + if !ok { + return &squad.ExecResult{Success: false, ErrorReason: "project workflow source guard unavailable"}, nil + } + wsDir := filepath.Dir(filepath.Dir(step.WorkDir)) + strict := step.IntentID == IntentTool || step.IntentID == IntentArtifact || step.IntentID == IntentScript || step.IntentID == IntentCritic + if err := verifyPriorWorkflowOutputs(ctx, guard, wsDir, r.subOf, r.stepKinds, strict); err != nil { + return &squad.ExecResult{Success: false, ErrorReason: err.Error()}, nil + } + } var content string var fields map[string]string var err error diff --git a/internal/wfcompile/project_backend.go b/internal/wfcompile/project_backend.go index 47e2e1c..08eb01a 100644 --- a/internal/wfcompile/project_backend.go +++ b/internal/wfcompile/project_backend.go @@ -2,6 +2,7 @@ package wfcompile import ( "context" + "path/filepath" "agent-squad/squad" "agent-system-backend/internal/projectcontext" @@ -10,7 +11,7 @@ import ( // 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 { +func projectStepBackends(base *squad.BackendResolver, rc ProjectStepProvenance, subOf, kinds map[string]string) *squad.BackendResolver { wrapped := squad.NewBackendResolver() for _, name := range base.ListBackends() { backendName := name @@ -19,13 +20,17 @@ func projectStepBackends(base *squad.BackendResolver) *squad.BackendResolver { if err != nil { return nil, err } - return &projectStepExecutor{next: backend}, nil + return &projectStepExecutor{next: backend, rc: rc, subOf: subOf, kinds: kinds}, nil }) } return wrapped } -type projectStepExecutor struct{ next squad.Executor } +type projectStepExecutor struct { + next squad.Executor + rc ProjectStepProvenance + subOf, kinds map[string]string +} func (e *projectStepExecutor) Name() string { return e.next.Name() } @@ -33,5 +38,11 @@ func (e *projectStepExecutor) Execute(ctx context.Context, req squad.ExecRequest if req.ExecutionID == "" { return &squad.ExecResult{Success: false, ErrorReason: "project workflow step attempt unavailable"}, nil } + if e.rc != nil { + wsDir := filepath.Dir(filepath.Dir(req.WorkDir)) + if err := verifyPriorWorkflowOutputs(ctx, e.rc, wsDir, e.subOf, e.kinds, true); err != nil { + return &squad.ExecResult{Success: false, ErrorReason: err.Error()}, 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 index 0c971ab..c191f94 100644 --- a/internal/wfcompile/project_sources.go +++ b/internal/wfcompile/project_sources.go @@ -3,6 +3,7 @@ package wfcompile import ( "context" "errors" + "fmt" "os" "path/filepath" "sync" @@ -11,6 +12,41 @@ import ( "agent-squad/squad" ) +// verifyPriorWorkflowOutputs is shared by model and host steps. The engine's +// task brief/blackboard/handoff can carry earlier output as a fresh user +// prompt, so every completed machine step must have a current source proof. +func verifyPriorWorkflowOutputs(ctx context.Context, rc ProjectStepProvenance, wsDir string, subOf, kinds map[string]string, modelSend bool) error { + state, err := (squad.FileStore{}).LoadRun(wsDir) + if err != nil { + return fmt.Errorf("project workflow state unavailable: %w", err) + } + for _, step := range state.Plan.Workflow.Steps { + if step.Status != "completed" { + continue + } + switch kinds[step.ID] { + case "prompt": + sub := subOf[step.ID] + if sub == "" || step.AttemptID == "" { + return fmt.Errorf("project workflow step %s lacks source identity", step.ID) + } + data, err := os.ReadFile(filepath.Join(wsDir, filepath.FromSlash(sub), "output.md")) + if err != nil { + return fmt.Errorf("project workflow step %s output unavailable: %w", step.ID, err) + } + if err := rc.VerifyProjectStepOutput(ctx, step.ID, step.AttemptID, "prompt", data); err != nil { + return fmt.Errorf("project workflow step %s output source changed: %w", step.ID, err) + } + case "hitl", "critic": // human input or a verdict whose prompt source was checked above + default: + if modelSend { + return fmt.Errorf("project workflow step %s has unproven machine output", step.ID) + } + } + } + return nil +} + // 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. diff --git a/internal/wfcompile/run.go b/internal/wfcompile/run.go index 8799e13..41c011b 100644 --- a/internal/wfcompile/run.go +++ b/internal/wfcompile/run.go @@ -80,6 +80,9 @@ func RunCompiled(rc Host, deps Deps, launch workflow.Launch, vars workflow.Vars, for _, step := range def.Steps { stepKinds[step.ID] = step.Kind } + for _, pair := range compiled.Critics { + stepKinds[pair.EvalID] = "critic" + } if rc.Run().ProjectID != "" { if _, ok := rc.(ProjectStepProvenance); !ok { return Outcome{}, fmt.Errorf("项目工作流缺少可信步骤产物绑定能力") @@ -97,7 +100,7 @@ func RunCompiled(rc Host, deps Deps, launch workflow.Launch, vars workflow.Vars, backends = squad.NewBackendResolver() } if rc.Run().ProjectID != "" { - backends = projectStepBackends(backends) + backends = projectStepBackends(backends, rc.(ProjectStepProvenance), compiled.SubOf, stepKinds) } runner := &hostRunner{ rc: rc, deps: deps, scripts: launch.Scripts, subOf: compiled.SubOf, diff --git a/internal/wfcompile/wfcompile_test.go b/internal/wfcompile/wfcompile_test.go index a9a57e2..7ec0bfa 100644 --- a/internal/wfcompile/wfcompile_test.go +++ b/internal/wfcompile/wfcompile_test.go @@ -476,7 +476,7 @@ 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"}, "") + backend, _, err := projectStepBackends(base, nil, nil, nil).Resolve(squad.RoleSpec{Role: "writer", Backend: "probe"}, "") if err != nil { t.Fatal(err) } @@ -563,11 +563,24 @@ func TestProjectToolCriticRejectsUnprovenOutputBeforeModel(t *testing.T) { 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(), "产物依据不可用") { + if err == nil || called || (!strings.Contains(err.Error(), "产物依据不可用") && !strings.Contains(err.Error(), "unproven machine output")) { t.Fatalf("unproven tool output reached judge: err=%v called=%v", err, called) } } +func TestProjectModelStepRejectsEarlierToolOutputInTaskBrief(t *testing.T) { + f := &projectSourceHost{fakeHost: &fakeHost{toolNames: []string{"time_now"}}} + be := &seqBackend{texts: []string{"should not run"}} + def := core.WorkflowDef{ID: "tool-then-model", Title: "Tool then model", Steps: []core.WorkflowStep{ + {ID: "get", Title: "Read", Kind: "tool", Tool: "time_now"}, + {ID: "draft", Title: "Draft", Kind: "prompt", Agent: "writer", Prompt: "use {{steps.get.output}}"}, + }} + _, err := RunCompiled(f, Deps{DataDir: t.TempDir(), Backends: seqResolver(t, be), DefaultBackend: "seq"}, workflow.Launch{Def: def}, workflow.Vars{Input: "write"}, RunOptions{}) + if err == nil || be.calls != 0 || !strings.Contains(err.Error(), "unproven machine output") { + t.Fatalf("tool output entered project model step: err=%v modelCalls=%d", err, be.calls) + } +} + // 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 2010b0c..9c207bb 100644 --- a/pkg/extension/project_context.go +++ b/pkg/extension/project_context.go @@ -83,6 +83,27 @@ type ProjectContextProvider interface { ReadEvidence(context.Context, ProjectEvidenceRequest) (ProjectEvidence, error) } +// ProjectContextSourceRecheckRequest names only sources already present in a +// trusted model-input receipt. Scope identity is rebuilt by Backend from the +// persisted run; no prompt, output body, or caller-chosen owner is accepted. +type ProjectContextSourceRecheckRequest struct { + Scope ProjectContextRequest + Included []ProjectContextManifestEntry + OfferedReferences []ProjectEvidenceRef + ReadEvidence []ProjectContextManifestEntry +} + +// ProjectContextSourceRecheckResult contains current authorized source text +// only for transient Backend hash verification. It is never stored as a new +// model receipt. Every requested source/ref must be present in Bundle. +type ProjectContextSourceRecheckResult struct{ Bundle ProjectContextBundle } + +// ProjectContextSourceRechecker is an optional method of the same registered +// ProjectContextProvider. Project workflow reuse fails closed without it. +type ProjectContextSourceRechecker interface { + Recheck(context.Context, ProjectContextSourceRecheckRequest) (ProjectContextSourceRecheckResult, error) +} + // WorkCompletionRequest gives the generation goal and optional target. It has // no owner/project field; both are resolved from the dispatch host's stored // conversation immediately before the model call. diff --git a/pkg/extension/surface.txt b/pkg/extension/surface.txt index c3a08ec..e8d69ad 100644 --- a/pkg/extension/surface.txt +++ b/pkg/extension/surface.txt @@ -46,6 +46,10 @@ ProjectContextReady ProjectContextRegistry ProjectContextRegistry.RegisterProjectContextProvider ProjectContextRequest +ProjectContextSourceRecheckRequest +ProjectContextSourceRecheckResult +ProjectContextSourceRechecker +ProjectContextSourceRechecker.Recheck ProjectContextStatus ProjectEvidence ProjectEvidenceRef