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