Recheck project workflow sources without storing prompts

Why: Goal-sensitive workflow verification persisted model prompts and prior output text in trusted receipt rows, while cross-step model inputs could reuse stale machine output.

Evidence: RED receipt privacy test; full Go suite, targeted race, isolated PostgreSQL store/runmgr/storage-migrate tests, vet, and static build passed locally before commit. A known unrelated TempDir cleanup flake failed once and passed on targeted and serial full replay.

Impact: Add optional same-provider source recheck, verify every included item and evidence body before reuse, gate cross-step outputs, reject unsettled attempts, remove raw scope columns, scrub isolated candidate v34 SQLite/PG schemas, and reject unsafe backups.

Refs: Approved RDM stage6 W6 source-recheck design adjustment r1; W6 independent review.
This commit is contained in:
weichen
2026-10-14 05:38:11 +00:00
parent 6ba53e43ed
commit ecb86f4309
26 changed files with 884 additions and 104 deletions
+43
View File
@@ -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 测试同隔离约定)。
+23
View File
@@ -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 不同(换了快照续传)→ 拒绝;
+53
View File
@@ -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")
}
}
+1
View File
@@ -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)
+27
View File
@@ -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
+42 -2
View File
@@ -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 {
+8 -2
View File
@@ -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)
+9 -2
View File
@@ -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")
}
}
+142 -52
View File
@@ -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")
+154 -2
View File
@@ -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
})
+1
View File
@@ -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 整文档单块)
+8 -8
View File
@@ -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)
+1 -1
View File
@@ -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)
+8 -11
View File
@@ -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)
}
+130 -4
View File
@@ -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")
}
+68 -5
View File
@@ -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
+1
View File
@@ -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 版本。
+44 -8
View File
@@ -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
}
+16 -1
View File
@@ -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)
}
+11
View File
@@ -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
+14 -3
View File
@@ -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)
}
+36
View File
@@ -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.
+4 -1
View File
@@ -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,
+15 -2
View File
@@ -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)
+21
View File
@@ -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.
+4
View File
@@ -46,6 +46,10 @@ ProjectContextReady
ProjectContextRegistry
ProjectContextRegistry.RegisterProjectContextProvider
ProjectContextRequest
ProjectContextSourceRecheckRequest
ProjectContextSourceRecheckResult
ProjectContextSourceRechecker
ProjectContextSourceRechecker.Recheck
ProjectContextStatus
ProjectEvidence
ProjectEvidenceRef