489 lines
19 KiB
Go
489 lines
19 KiB
Go
package runmgr
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"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 (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
|
|
}
|
|
|
|
// 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
|
|
}
|
|
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
|
|
}
|
|
if err := ctx.Err(); 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
|
|
}
|
|
// Set-wise, not provider-order-wise: a benign reordering of the same
|
|
// recorded identities is tolerated, while any added, dropped or replaced
|
|
// entry still fails closed. Order carries no authorization meaning.
|
|
if len(current.Items) != len(m.Included) || len(current.Evidence) != len(expectedRefs) || !sameMissingSet(current.Missing, m.Missing) {
|
|
return errors.New("project source recheck returned an incomplete source set")
|
|
}
|
|
// Each returned item must account for exactly one recorded input; the hash
|
|
// is recomputed here, never taken from the provider's echo of the request.
|
|
claimed := make([]bool, len(m.Included))
|
|
for _, item := range current.Items {
|
|
matched := false
|
|
for i, used := range m.Included {
|
|
if claimed[i] || item.SourceID != used.SourceID || item.Revision != used.Revision || item.Hash != used.SourceHash {
|
|
continue
|
|
}
|
|
if fmt.Sprintf("%x", sha256.Sum256([]byte(item.Text))) != used.InputSHA256 {
|
|
return fmt.Errorf("project context source %q text changed under its recorded identity", used.SourceID)
|
|
}
|
|
claimed[i], matched = true, true
|
|
break
|
|
}
|
|
if !matched {
|
|
return errors.New("project source recheck returned an unrelated or duplicate source")
|
|
}
|
|
}
|
|
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")
|
|
}
|
|
}
|
|
// Load-bearing fail-closed close: a context cancelled or expired after the
|
|
// last structural pass above must still reject. Do not relax this to
|
|
// `return nil`; a late cancellation would otherwise admit machine context
|
|
// that was never rechecked against live authorization.
|
|
return ctx.Err()
|
|
}
|
|
|
|
// sameMissingSet compares the rechecked and recorded missing-source lists as
|
|
// sets. Neither input is mutated; any difference other than order is a
|
|
// rejection, because a withdrawn or newly missing source changes what the
|
|
// original model request actually saw.
|
|
func sameMissingSet(current, recorded []string) bool {
|
|
if len(current) != len(recorded) {
|
|
return false
|
|
}
|
|
left, right := slices.Clone(current), slices.Clone(recorded)
|
|
slices.Sort(left)
|
|
slices.Sort(right)
|
|
return slices.Equal(left, right)
|
|
}
|
|
|
|
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
|
|
}
|
|
req, err := c.trustedProjectContextRequest(ctx, budget)
|
|
if err != nil {
|
|
return extension.ProjectContextBundle{}, true, err
|
|
}
|
|
req.Goal = goal
|
|
bundle, err := c.mgr.projectContextResolver(ctx, req)
|
|
if err != nil {
|
|
return extension.ProjectContextBundle{}, true, err
|
|
}
|
|
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 {
|
|
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{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 := 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")
|
|
}
|
|
if receipt.AttemptID == bound.AttemptID {
|
|
matchedBound = true
|
|
}
|
|
if err := c.recheckManifest(ctx, receipt, stepID); err != nil {
|
|
return fmt.Errorf("project workflow output source changed: %w", err)
|
|
}
|
|
}
|
|
if !matchedBound {
|
|
return errors.New("project workflow output binding lost its responded model attempt")
|
|
}
|
|
return nil
|
|
}
|