585 lines
31 KiB
Go
585 lines
31 KiB
Go
package runmgr
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"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"
|
|
)
|
|
|
|
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)
|
|
conv := recoveryConversation(t, st)
|
|
run := recoveryRun(t, st, conv, core.RunPending)
|
|
allowed := true
|
|
m.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return allowed, nil })
|
|
called := 0
|
|
m.SetProjectContextResolver(func(_ context.Context, req extension.ProjectContextRequest) (extension.ProjectContextBundle, error) {
|
|
called++
|
|
if req.Actor != string(run.Owner) || req.ProjectID != run.ProjectID || req.ConversationID != run.ConversationID || req.RunID != run.ID {
|
|
t.Errorf("context scope did not come from persisted run: %+v", req)
|
|
}
|
|
return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID}, nil
|
|
})
|
|
rc := &runCtx{mgr: m, run: run, ctx: context.Background()}
|
|
if _, active, err := rc.ProjectContext(context.Background(), "work goal", 900); err != nil || !active || called != 1 {
|
|
t.Fatalf("valid persisted scope rejected: active=%v called=%d err=%v", active, called, err)
|
|
}
|
|
allowed = false
|
|
if _, _, err := rc.ProjectContext(context.Background(), "work goal", 900); err == nil || called != 1 {
|
|
t.Fatal("revoked project still reached context provider")
|
|
}
|
|
allowed = true
|
|
rc.run.ProjectID = "forged"
|
|
if _, _, err := rc.ProjectContext(context.Background(), "work goal", 900); err == nil || called != 1 {
|
|
t.Fatal("in-memory project override reached context provider")
|
|
}
|
|
}
|
|
|
|
func TestProjectContextResumeReloadsCurrentVersionAndAuthorization(t *testing.T) {
|
|
st := openRecoveryStore(t)
|
|
m := newRecoveryManager(t, st)
|
|
conv := recoveryConversation(t, st)
|
|
run := recoveryRun(t, st, conv, core.RunInterrupted)
|
|
allowed, version := true, "v1"
|
|
m.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return allowed, nil })
|
|
m.SetProjectContextResolver(func(_ context.Context, req extension.ProjectContextRequest) (extension.ProjectContextBundle, error) {
|
|
return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID, Version: version}, nil
|
|
})
|
|
first := &runCtx{mgr: m, run: run, ctx: context.Background()}
|
|
if b, _, err := first.ProjectContext(context.Background(), "work", 900); err != nil || b.Version != "v1" {
|
|
t.Fatalf("initial context: %+v err=%v", b, err)
|
|
}
|
|
reloaded, err := st.GetRunByID(string(run.ID))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
resumed := &runCtx{mgr: m, run: reloaded, ctx: context.Background()}
|
|
version = "v2"
|
|
if b, _, err := resumed.ProjectContext(context.Background(), "work", 900); err != nil || b.Version != "v2" {
|
|
t.Fatalf("resumed context reused stale version: %+v err=%v", b, err)
|
|
}
|
|
allowed = false
|
|
if _, _, err := resumed.ProjectContext(context.Background(), "work", 900); err == nil {
|
|
t.Fatal("resumed context ignored current revocation")
|
|
}
|
|
}
|
|
|
|
func TestProjectJudgeChecksPersistedSourceEpochAcrossGuards(t *testing.T) {
|
|
st := openRecoveryStore(t)
|
|
m := newRecoveryManager(t, st)
|
|
conv := recoveryConversation(t, st)
|
|
run := recoveryRun(t, st, conv, core.RunPending)
|
|
m.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return true, nil })
|
|
epoch := "duty-1"
|
|
publication := "pub-1"
|
|
items := []extension.ProjectContextItem{{SourceID: "restricted-doc", Revision: "r1", Hash: "h1", Text: "OLD_RESTRICTED_BODY", Required: true}}
|
|
m.SetProjectContextResolver(func(_ context.Context, req extension.ProjectContextRequest) (extension.ProjectContextBundle, error) {
|
|
return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID, AuthorizationEpoch: epoch, PublicationEpoch: publication, Watermark: "commit-1", Items: items}, nil
|
|
})
|
|
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)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, manifest, err := projectcontext.Assemble(bundle, 900)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
manifest.InputSHA256, manifest.AttemptID, manifest.Delivery = "input-hash", "step-attempt", "attempted"
|
|
manifest.WorkflowStepAttemptID = "engine-run/draft/1"
|
|
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(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 {
|
|
t.Fatal(err)
|
|
}
|
|
if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err != nil {
|
|
t.Fatalf("same epoch output rejected: %v", err)
|
|
}
|
|
if err := rc.ValidateProjectCrossGuard(context.Background(), bundle); err != nil {
|
|
t.Fatalf("same epoch cross-guard request rejected: %v", err)
|
|
}
|
|
if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("replaced bytes")); err == nil {
|
|
t.Fatal("replaced output accepted")
|
|
}
|
|
if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/2", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil {
|
|
t.Fatal("second execution reused first attempt's identical output")
|
|
}
|
|
if err := rc.VerifyProjectStepOutput(context.Background(), "other", "engine-run/other/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil {
|
|
t.Fatal("another step reused output binding")
|
|
}
|
|
items = nil // provider withdraws the document while retaining project membership and epoch
|
|
if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil {
|
|
t.Fatal("withdrawn document accepted with unchanged authorization epoch")
|
|
}
|
|
items = []extension.ProjectContextItem{{SourceID: "restricted-doc", Revision: "r1", Hash: "h1", Text: "OLD_RESTRICTED_BODY", Required: true}}
|
|
epoch = "duty-2"
|
|
current, _, err := rc.ProjectContext(context.Background(), "judge", 900)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := rc.ValidateProjectCrossGuard(context.Background(), current); err == nil {
|
|
t.Fatal("cross-guard request ignored persisted epoch change")
|
|
}
|
|
if err := rc.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil {
|
|
t.Fatal("revoked source epoch accepted for workflow judge")
|
|
}
|
|
reloaded, err := st.GetRunByID(string(run.ID))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
resumed := &runCtx{mgr: m, run: reloaded, ctx: context.Background()}
|
|
if err := resumed.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil {
|
|
t.Fatal("resumed judge ignored revocation")
|
|
}
|
|
epoch, publication = "duty-1", "pub-2"
|
|
if err := resumed.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("OLD_RESTRICTED_BODY")); err == nil {
|
|
t.Fatal("resumed judge ignored publication epoch change")
|
|
}
|
|
other := recoveryRun(t, st, conv, core.RunPending)
|
|
noReceipt := &runCtx{mgr: m, run: other, ctx: context.Background()}
|
|
if err := noReceipt.BindProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", started, []byte("tool body")); err == nil {
|
|
t.Fatal("tool-only output without model provenance accepted")
|
|
}
|
|
if err := noReceipt.VerifyProjectStepOutput(context.Background(), "draft", "engine-run/draft/1", "prompt", []byte("tool body")); err == nil {
|
|
t.Fatal("another run reused output binding")
|
|
}
|
|
}
|
|
|
|
func 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
|
|
})
|
|
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)
|
|
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
|
|
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
|
|
}
|
|
}
|
|
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 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)
|
|
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
|
|
})
|
|
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)
|
|
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)
|
|
}
|
|
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)
|
|
}
|
|
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)
|
|
conv := recoveryConversation(t, st)
|
|
run := recoveryRun(t, st, conv, core.RunPending)
|
|
rc := &runCtx{mgr: m, run: run, ctx: context.Background()}
|
|
manifest := extension.ProjectContextManifest{ProjectID: run.ProjectID, Version: "v3", Model: "model-1", AttemptID: "attempt-1", Delivery: "attempted", InputSHA256: "abc", Included: []extension.ProjectContextManifestEntry{{SourceID: "s1", Revision: "r2", SourceHash: "h2", InputSHA256: "def"}}}
|
|
if err := rc.SaveContextManifest(context.Background(), manifest); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
got, err := st.ListContextManifests(string(run.Owner), string(run.ID))
|
|
if err != nil || len(got) != 1 || got[0].Version != "v3" || got[0].Included[0].SourceID != "s1" {
|
|
t.Fatalf("manifest missing from persisted run audit: got=%+v err=%v", got, err)
|
|
}
|
|
other, err := st.ListContextManifests("other", string(run.ID))
|
|
if err != nil || len(other) != 0 {
|
|
t.Fatalf("cross-owner manifest visible: got=%+v err=%v", other, err)
|
|
}
|
|
}
|
|
|
|
func TestContextManifestRejectsDocumentDutyRevocationBeforeSend(t *testing.T) {
|
|
st := openRecoveryStore(t)
|
|
m := newRecoveryManager(t, st)
|
|
conv := recoveryConversation(t, st)
|
|
run := recoveryRun(t, st, conv, core.RunPending)
|
|
m.SetProjectWorkAuthorization(func(context.Context, string, core.ID) (bool, error) { return true, nil })
|
|
epoch := "duty-1"
|
|
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", 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)
|
|
if err != nil || !active {
|
|
t.Fatalf("initial context: active=%v err=%v", active, err)
|
|
}
|
|
_, manifest, err := projectcontext.Assemble(bundle, 900)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
manifest.InputSHA256, manifest.AttemptID, manifest.Delivery = "request-hash", "attempt-revoked", "attempted"
|
|
epoch = "duty-2" // project work remains allowed; document duty changed
|
|
if err := rc.SaveContextManifest(context.Background(), manifest); err == nil {
|
|
t.Fatal("stale document duty accepted at model send")
|
|
}
|
|
stored, err := st.ListContextManifests(string(run.Owner), string(run.ID))
|
|
if err != nil || len(stored) != 0 {
|
|
t.Fatalf("revoked context wrote a send receipt: %+v err=%v", stored, err)
|
|
}
|
|
}
|
|
|
|
func TestProjectEvidenceOnlyReadsCurrentBundleReference(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 })
|
|
listed := true
|
|
m.SetProjectContextResolver(func(_ context.Context, req extension.ProjectContextRequest) (extension.ProjectContextBundle, error) {
|
|
b := extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: req.ProjectID}
|
|
if listed {
|
|
b.Evidence = []extension.ProjectEvidenceRef{{ID: "ref-1", SourceID: "doc-1", Revision: "r3", Hash: "source-hash"}}
|
|
}
|
|
return b, nil
|
|
})
|
|
read := 0
|
|
m.SetProjectEvidenceReader(func(_ context.Context, req extension.ProjectEvidenceRequest) (extension.ProjectEvidence, error) {
|
|
read++
|
|
if req.Scope.RunID != run.ID || req.Reference.ID != "ref-1" {
|
|
t.Errorf("untrusted evidence request: %+v", req)
|
|
}
|
|
return extension.ProjectEvidence{Reference: req.Reference, Text: "authorized detail"}, nil
|
|
})
|
|
rc := &runCtx{mgr: m, run: run, ctx: context.Background()}
|
|
if _, err := rc.ProjectEvidence(context.Background(), "forged-ref"); err == nil || read != 0 {
|
|
t.Fatal("arbitrary reference reached provider")
|
|
}
|
|
evidence, err := rc.ProjectEvidence(context.Background(), "ref-1")
|
|
if err != nil || evidence.Text != "authorized detail" || read != 1 {
|
|
t.Fatalf("current reference rejected: evidence=%+v read=%d err=%v", evidence, read, err)
|
|
}
|
|
if got := rc.ContextEvidenceInRequest([]agentllm.Message{{Role: agentllm.RoleTool, Content: "authorized detail"}}); len(got) != 1 || got[0].SourceID != "doc-1" {
|
|
t.Fatalf("actual evidence read missing from request manifest: %+v", got)
|
|
}
|
|
if got := rc.ContextEvidenceInRequest([]agentllm.Message{{Role: agentllm.RoleTool, Content: "cleared old tool result"}}); len(got) != 0 {
|
|
t.Fatalf("cleared evidence still counted as model input: %+v", got)
|
|
}
|
|
listed = false
|
|
if _, err := rc.ProjectEvidence(context.Background(), "ref-1"); err == nil || read != 1 {
|
|
t.Fatal("revoked reference reached provider")
|
|
}
|
|
if err := rc.SaveContextManifest(context.Background(), extension.ProjectContextManifest{ProjectID: run.ProjectID, InputSHA256: "new-request-hash", AttemptID: "attempt-revoked", Delivery: "attempted"}); err == nil {
|
|
t.Fatal("new model request accepted after an earlier evidence read was revoked")
|
|
}
|
|
}
|
|
|
|
func TestConcurrentProjectEvidenceReadsKeepTrustedScope(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 })
|
|
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
|
|
})
|
|
m.SetProjectEvidenceReader(func(_ context.Context, req extension.ProjectEvidenceRequest) (extension.ProjectEvidence, error) {
|
|
return extension.ProjectEvidence{Reference: req.Reference, Text: "body"}, nil
|
|
})
|
|
rc := &runCtx{mgr: m, run: run, ctx: context.Background()}
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < 8; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
if _, err := rc.ProjectEvidence(context.Background(), "r1"); err != nil {
|
|
t.Error(err)
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
}
|