weichen
6ba53e43ed
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.
356 lines
15 KiB
Go
356 lines
15 KiB
Go
package runmgr
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
agentllm "agent-native/runtime/llm"
|
|
|
|
"agent-system-backend/internal/projectcontext"
|
|
"agent-system-backend/internal/store"
|
|
"agent-system-backend/pkg/extension"
|
|
core "agent-system-core"
|
|
)
|
|
|
|
// SetProjectContextResolver activates project-context reads for persisted runs.
|
|
// Nil preserves the existing non-project and pre-provider behavior.
|
|
func (m *Manager) SetProjectContextResolver(fn func(context.Context, extension.ProjectContextRequest) (extension.ProjectContextBundle, error)) {
|
|
m.projectContextResolver = fn
|
|
}
|
|
|
|
func (m *Manager) SetProjectEvidenceReader(fn func(context.Context, extension.ProjectEvidenceRequest) (extension.ProjectEvidence, error)) {
|
|
m.projectEvidenceReader = fn
|
|
}
|
|
|
|
func (c *runCtx) ProjectContextEnabled() bool {
|
|
return c.run.ProjectID != "" && c.mgr.projectContextResolver != nil
|
|
}
|
|
|
|
// ValidateProjectCrossGuard links separate model streams in one persisted run.
|
|
// A new guard cannot treat a workflow/user prompt as fresh when earlier model
|
|
// work under the same run used another authorization or publication revision.
|
|
func (c *runCtx) ValidateProjectCrossGuard(ctx context.Context, current extension.ProjectContextBundle) error {
|
|
if !c.ProjectContextEnabled() {
|
|
return nil
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
prior, err := c.mgr.st.ListContextManifests(string(c.run.Owner), string(c.run.ID))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, m := range prior {
|
|
if m.ProjectID != current.ProjectID || m.AuthorizationEpoch == "" || m.PublicationEpoch == "" ||
|
|
current.AuthorizationEpoch == "" || current.PublicationEpoch == "" ||
|
|
m.AuthorizationEpoch != current.AuthorizationEpoch || m.PublicationEpoch != current.PublicationEpoch || m.Watermark != current.Watermark {
|
|
return errors.New("prior project model work used a different authorization or publication revision")
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *runCtx) contextRequest() extension.ProjectContextRequest {
|
|
c.projectContextMu.Lock()
|
|
defer c.projectContextMu.Unlock()
|
|
return c.projectContextReq
|
|
}
|
|
|
|
func (c *runCtx) setContextRequest(req extension.ProjectContextRequest) {
|
|
c.projectContextMu.Lock()
|
|
c.projectContextReq = req
|
|
c.projectContextMu.Unlock()
|
|
}
|
|
|
|
type projectEvidenceRecord struct {
|
|
ref extension.ProjectEvidenceRef
|
|
text string
|
|
}
|
|
|
|
// ContextEvidenceInRequest reports only details still present in the actual
|
|
// model request after transcript compaction and window cleanup.
|
|
func (c *runCtx) ContextEvidenceInRequest(messages []agentllm.Message) []extension.ProjectContextManifestEntry {
|
|
c.projectEvidenceMu.Lock()
|
|
defer c.projectEvidenceMu.Unlock()
|
|
var out []extension.ProjectContextManifestEntry
|
|
for _, used := range c.projectEvidenceUsed {
|
|
for _, msg := range messages {
|
|
if !strings.Contains(msg.Content, used.text) {
|
|
continue
|
|
}
|
|
out = append(out, extension.ProjectContextManifestEntry{
|
|
ReferenceID: used.ref.ID,
|
|
SourceID: used.ref.SourceID, Revision: used.ref.Revision, SourceHash: used.ref.Hash,
|
|
InputSHA256: fmt.Sprintf("%x", sha256.Sum256([]byte(used.text))),
|
|
})
|
|
break
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// ProjectEvidence accepts only a stable reference listed by a fresh Resolve.
|
|
// The provider must reauthorize the selected resource when reading its body.
|
|
func (c *runCtx) ProjectEvidence(ctx context.Context, id string) (extension.ProjectEvidence, error) {
|
|
if id == "" || c.mgr.projectEvidenceReader == nil || !c.ProjectContextEnabled() {
|
|
return extension.ProjectEvidence{}, errors.New("project evidence unavailable")
|
|
}
|
|
prior := c.contextRequest()
|
|
budget := prior.BudgetChars
|
|
if budget <= 0 {
|
|
budget = 12000
|
|
}
|
|
bundle, active, err := c.ProjectContext(ctx, prior.Goal, budget)
|
|
if err != nil || !active {
|
|
return extension.ProjectEvidence{}, errors.New("project evidence scope unavailable")
|
|
}
|
|
for _, ref := range bundle.Evidence {
|
|
if ref.ID != id {
|
|
continue
|
|
}
|
|
scope := c.contextRequest()
|
|
scope.BudgetChars = 8000
|
|
evidence, err := c.mgr.projectEvidenceReader(ctx, extension.ProjectEvidenceRequest{Scope: scope, Reference: ref})
|
|
if err != nil {
|
|
return extension.ProjectEvidence{}, err
|
|
}
|
|
if evidence.Reference != ref || strings.TrimSpace(evidence.Text) == "" || len([]rune(evidence.Text)) > 8000 {
|
|
return extension.ProjectEvidence{}, errors.New("project evidence response does not match authorized reference or budget")
|
|
}
|
|
c.projectEvidenceMu.Lock()
|
|
c.projectEvidenceUsed = append(c.projectEvidenceUsed, projectEvidenceRecord{ref: ref, text: evidence.Text})
|
|
c.projectEvidenceMu.Unlock()
|
|
return evidence, nil
|
|
}
|
|
return extension.ProjectEvidence{}, errors.New("project evidence reference not in current bundle")
|
|
}
|
|
|
|
// ProjectContext reloads identity and project from trusted run and conversation
|
|
// rows on every model request, including a resumed or background run.
|
|
func (c *runCtx) ProjectContext(ctx context.Context, goal string, budget int) (extension.ProjectContextBundle, bool, error) {
|
|
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))
|
|
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,
|
|
}
|
|
bundle, err := c.mgr.projectContextResolver(ctx, req)
|
|
if err != nil {
|
|
return extension.ProjectContextBundle{}, true, err
|
|
}
|
|
if bundle.ProjectID != run.ProjectID || bundle.Status != extension.ProjectContextReady {
|
|
return extension.ProjectContextBundle{}, true, errors.New("project context unavailable for trusted run")
|
|
}
|
|
c.setContextRequest(req)
|
|
return bundle, true, 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 {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
run, err := c.mgr.st.GetRunByID(string(c.run.ID))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
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" {
|
|
if err := c.mgr.authorizeProjectWork(ctx, string(run.Owner), core.ID(run.ProjectID)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
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")
|
|
}
|
|
current, err := c.mgr.projectContextResolver(ctx, resolvedReq)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := projectcontext.ValidateCurrent(m, current); err != nil {
|
|
return err
|
|
}
|
|
c.projectEvidenceMu.Lock()
|
|
pastReads := append([]projectEvidenceRecord(nil), c.projectEvidenceUsed...)
|
|
c.projectEvidenceMu.Unlock()
|
|
for _, used := range pastReads {
|
|
stillListed := false
|
|
for _, ref := range current.Evidence {
|
|
if ref == used.ref {
|
|
stillListed = true
|
|
break
|
|
}
|
|
}
|
|
if !stillListed || c.mgr.projectEvidenceReader == nil {
|
|
return errors.New("previous project evidence no longer authorized")
|
|
}
|
|
scope := resolvedReq
|
|
scope.BudgetChars = 8000
|
|
now, err := c.mgr.projectEvidenceReader(ctx, extension.ProjectEvidenceRequest{Scope: scope, Reference: used.ref})
|
|
if err != nil || now.Reference != used.ref || now.Text != used.text {
|
|
return errors.New("previous project evidence changed before model send")
|
|
}
|
|
}
|
|
for _, used := range m.ReadEvidence {
|
|
if c.mgr.projectEvidenceReader == nil {
|
|
return errors.New("project evidence reader missing at model send")
|
|
}
|
|
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 := resolvedReq
|
|
scope.BudgetChars = 8000
|
|
evidence, err := c.mgr.projectEvidenceReader(ctx, extension.ProjectEvidenceRequest{Scope: scope, Reference: ref})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if evidence.Reference == ref && fmt.Sprintf("%x", sha256.Sum256([]byte(evidence.Text))) == used.InputSHA256 {
|
|
matched = true
|
|
break
|
|
}
|
|
}
|
|
if !matched {
|
|
return errors.New("project evidence changed before model send")
|
|
}
|
|
}
|
|
}
|
|
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)
|
|
}
|
|
|
|
// InvalidateProjectStepOutput prevents a retry from inheriting the prior
|
|
// attempt's binding, even when the new attempt produces identical bytes.
|
|
func (c *runCtx) InvalidateProjectStepOutput(ctx context.Context, stepID string) error {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
return c.mgr.st.UnbindWorkflowStepSource(string(c.run.Owner), string(c.run.ID), stepID)
|
|
}
|
|
|
|
// BindProjectStepOutput stores exact bytes and their responded model attempt
|
|
// after a prompt step has materialized. The step start timestamp fences out
|
|
// receipts from earlier steps; the workflow sink rejects concurrent ambiguity.
|
|
func (c *runCtx) BindProjectStepOutput(ctx context.Context, stepID, stepAttemptID, kind string, startedAt time.Time, output []byte) error {
|
|
if kind != "prompt" || !c.ProjectContextEnabled() {
|
|
return errors.New("project prompt step provenance unavailable")
|
|
}
|
|
_, err := c.mgr.st.BindWorkflowStepSource(string(c.run.Owner), string(c.run.ID), stepID, stepAttemptID, kind, startedAt, output)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
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
|
|
}
|
|
|
|
// VerifyProjectStepOutput is the last gate before a workflow critic receives
|
|
// output.md. It checks the persisted run+step association, current file bytes,
|
|
// the trusted responded request, and live domain authorization epochs.
|
|
func (c *runCtx) VerifyProjectStepOutput(ctx context.Context, stepID, stepAttemptID, kind string, output []byte) error {
|
|
if kind != "prompt" || !c.ProjectContextEnabled() {
|
|
return errors.New("project workflow output has no verifiable prompt source")
|
|
}
|
|
bound, err := c.mgr.st.WorkflowStepSource(string(c.run.Owner), string(c.run.ID), stepID)
|
|
if err != nil {
|
|
return errors.New("project workflow output has no trusted step binding")
|
|
}
|
|
if bound.ProjectID != string(c.run.ProjectID) || bound.StepAttemptID != stepAttemptID || bound.OutputSHA256 != sha256.Sum256(output) {
|
|
return errors.New("project workflow output bytes or project scope changed")
|
|
}
|
|
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")
|
|
}
|
|
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")
|
|
}
|
|
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")
|
|
}
|
|
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")
|
|
}
|
|
return nil
|
|
}
|