Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion docs/architecture/governance.md
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,9 @@ child prompt-injection defense.
`Service.admitMode` (`internal/adapter/server/service.go`) refuses a created or switched
mode above the ceiling, and `auto` without an enforcing guardrails checker; a server
whose default mode is `auto` refuses to start without one. Unchecked autonomy is the
explicit `yolo` choice, where a configured checker is advisory. `admitStoredMode`
explicit `yolo` choice, where a configured checker never blocks: the engine skips it
unless the reviewer opts in (`YoloReviewPolicyProvider`), and then reviews off the
dispatch path, cancelling reviews still in flight when the run ends. `admitStoredMode`
re-checks a persisted session's mode at run, approval resume, fork, and clear, so a
mode admitted under a looser deployment fails with `FailedPrecondition` until the owner
switches it; history is untouched. A session parked awaiting an approval in such a mode
Expand Down
21 changes: 21 additions & 0 deletions engine/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,18 @@ The covered surface is the eight core packages (`session`, `governance`, `learni

### Added

- **Yolo advisory review opt-in** — adds `agent.YoloReviewPolicyProvider`. A
`ToolReviewer` that implements it and returns true from `GuardrailYoloReviews`
has its applicable action and inbound reviews in yolo-mode sessions run in the
background: the tool call neither waits for the verdict nor fails closed, and
the verdict is emitted as an advisory guardrail `EvHook` after the tool result
and before the run's terminal event. At most four run at once; calls beyond
that are not reviewed. Run end cancels reviews still in flight and drops their
verdicts. Advisory verdicts are not recorded in the review trajectory. Adds
`ReviewEvidencePreparation.Snapshot`, set for these reviews, asking the
preparer to capture workspace evidence at preparation time because the
reviewed call may change it before the checker reads it. Added (minor).

- **Provider stop detail** — adds `port.Chunk.StopDetail`, the provider's own
terminal label (for example `max_tokens`, `refusal`, `length`) that an adapter
collapsed into `StopError` on `ChunkDone`. A stop-only `StopError` run now names
Expand Down Expand Up @@ -455,6 +467,15 @@ The covered surface is the eight core packages (`session`, `governance`, `learni

### Changed

- **Yolo-mode sessions skip contextual review by default** — a yolo-mode session
no longer calls the `ToolReviewer` for action or inbound review unless the
reviewer implements `YoloReviewPolicyProvider` and opts in. Previously review
ran inline and the tool waited for an advisory verdict. A terminal review
failure (for example evidence preparation) no longer fails closed in a yolo
session. Narrower delegated children of a yolo parent keep enforcing.
Changed (breaking for embedders that relied on inline yolo review; pre-v1
minor bump).

- **Agent-bound authority enforcement** — `Session.BindAuthority` requires a
ceiling when `AgentDefinitionName` is set and rejects tools outside that ceiling.
`GrantToolAuthority` and `CompleteWorkspaceEnrollment` reject tool grants beyond
Expand Down
17 changes: 9 additions & 8 deletions engine/agent/actionreview.go
Original file line number Diff line number Diff line change
Expand Up @@ -334,7 +334,7 @@ func (e *Engine) establishReviewPrincipal(r *Run) {
r.reviewRoot.establishInstructions(instructions)
}

func (e *Engine) prepareActionReview(ctx context.Context, r *Run, sess *session.Session, env tool.Environment, call session.ToolCall) actionReview {
func (e *Engine) prepareActionReview(ctx context.Context, r *Run, sess *session.Session, env tool.Environment, call session.ToolCall, snapshot bool) actionReview {
e.establishReviewPrincipal(r)
target, paths := reviewTarget(call)
facts, complete, principalRevision := r.reviewRoot.principalSnapshotWithRevision()
Expand Down Expand Up @@ -382,7 +382,7 @@ func (e *Engine) prepareActionReview(ctx context.Context, r *Run, sess *session.
var prepared PreparedReviewEvidence
var prepareErr error
if ctx.Err() == nil {
prepared, prepareErr = e.prepareReviewEvidence(ctx, r, sess, env, req, nil)
prepared, prepareErr = e.prepareReviewEvidence(ctx, r, sess, env, req, nil, snapshot)
} else {
prepareErr = reviewFailure(ReviewFailureTimeout)
}
Expand Down Expand Up @@ -428,12 +428,12 @@ func (e *Engine) authorizeReviewEvidenceRead(ctx context.Context, r *Run, sess *
return nil
}

func (e *Engine) prepareReviewEvidence(ctx context.Context, r *Run, sess *session.Session, env tool.Environment, req ToolReviewRequest, result *session.ToolResult) (PreparedReviewEvidence, error) {
func (e *Engine) prepareReviewEvidence(ctx context.Context, r *Run, sess *session.Session, env tool.Environment, req ToolReviewRequest, result *session.ToolResult, snapshot bool) (PreparedReviewEvidence, error) {
if r.reviewRoot.preparer == nil {
return PreparedReviewEvidence{}, nil
}
prepared, err := r.reviewRoot.preparer.PrepareReviewEvidence(ctx, ReviewEvidencePreparation{
Request: req, Environment: env, Owner: sess.Owner.Clone(), Result: result,
Request: req, Environment: env, Owner: sess.Owner.Clone(), Result: result, Snapshot: snapshot,
Authorize: func(authCtx context.Context, call session.ToolCall) error {
paths := tool.LocalFileOperands(call.Name, call.Args)
if call.Name != "Read" || len(paths) != 1 {
Expand Down Expand Up @@ -662,9 +662,9 @@ type actionReviewAssessment struct {
grantHit bool
}

func (e *Engine) prepareActionAssessment(ctx context.Context, r *Run, sess *session.Session, env tool.Environment, call session.ToolCall) actionReviewAssessment {
func (e *Engine) prepareActionAssessment(ctx context.Context, r *Run, sess *session.Session, env tool.Environment, call session.ToolCall, snapshot bool) actionReviewAssessment {
reviewCtx, cancel := context.WithTimeoutCause(ctx, reviewAssessmentTimeout, reviewFailure(ReviewFailureTimeout))
action := e.prepareActionReview(reviewCtx, r, sess, env, call)
action := e.prepareActionReview(reviewCtx, r, sess, env, call, snapshot)
assessment := actionReviewAssessment{action: action, ctx: reviewCtx, cancel: cancel}
if action.prepareErr == nil && reviewCtx.Err() == nil {
if issuer, ok := r.reviewRoot.reviewer.(ReviewGrantStore); ok && action.repeat {
Expand Down Expand Up @@ -944,9 +944,10 @@ func (e *Engine) reviewActionWithTail(ctx context.Context, r *Run, sess *session
e.publishToolResult(r, session.Event{Type: session.EvToolResult, Turn: turnIdx, ToolResult: ptr(res)})
return res, nil, false, false
}
e.startBackgroundActionReview(r, sess, env, turnIdx, call)
return session.ToolResult{}, nil, false, true
}
assessment := e.prepareActionAssessment(ctx, r, sess, env, call)
assessment := e.prepareActionAssessment(ctx, r, sess, env, call, false)
assessAction(ctx, r, &assessment)
if ctx.Err() != nil {
r.recordGuardrailUsageWhileActive(ctx, sess, assessment.usage)
Expand Down Expand Up @@ -990,7 +991,7 @@ func (e *Engine) armPostActionGrant(ctx context.Context, r *Run, sess *session.S
if !repeat || !hasIssuer || park != nil || cancelled || result.IsError {
return
}
post := e.prepareActionReview(ctx, r, sess, env, call)
post := e.prepareActionReview(ctx, r, sess, env, call, false)
if post.close != nil {
defer post.close()
}
Expand Down
196 changes: 196 additions & 0 deletions engine/agent/backgroundreview.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,196 @@
package agent

import (
"context"
"sync"
"time"

"github.com/stacklok/mecatl/engine/governance"
"github.com/stacklok/mecatl/engine/port"
"github.com/stacklok/mecatl/engine/session"
"github.com/stacklok/mecatl/engine/tool"
)

// backgroundReviewDrainGrace bounds how long run end waits for cancelled yolo
// advisory reviews to unwind. A var only as a TEST seam, like childDrainCap.
var backgroundReviewDrainGrace = 1 * time.Second

// maxBackgroundReviews caps concurrent advisory checker runs per run, bounding
// provider load, cost, and held evidence on read-heavy runs.
const maxBackgroundReviews = 4

// backgroundReviews is a run's set of opted-in yolo advisory reviews. Each
// review snapshots its request and evidence on the dispatching goroutine, then
// runs the checker off the dispatch path on a context detached from the run's
// own cancellation. At most maxBackgroundReviews run at once; a review over the
// cap is skipped. Advisory verdicts never feed the shared trajectory, so
// enforcing reviews of delegated children see the same facts in dispatch order. Run end cancels them; their advisory events go out through the
// child registry's sealed emit, so none follows the run's terminal event. The
// context is created on first use, so a run without background reviews pays
// no allocations for the set.
type backgroundReviews struct {
mu sync.Mutex
wg sync.WaitGroup
ctx context.Context //nolint:containedctx // run-scoped review lifetime, cancelled at run end
cancel context.CancelFunc
closed bool
active int
}

// reserve claims a review slot and returns the review context, creating it
// from parent on first use. It reports false once run end has begun draining
// or when maxBackgroundReviews are already running; a true return must be
// followed by start, which releases the slot.
func (b *backgroundReviews) reserve(parent context.Context) (context.Context, bool) {
b.mu.Lock()
defer b.mu.Unlock()
if b.closed || b.active >= maxBackgroundReviews {
return nil, false
}
b.active++
if b.ctx == nil {
if parent == nil {
parent = context.Background()
}
b.ctx, b.cancel = context.WithCancel(context.WithoutCancel(parent))
}
return b.ctx, true
}

// start runs a reserved review and releases its slot when it ends. It reports
// false, releasing the slot at once, when run end began after the reservation.
func (b *backgroundReviews) start(review func()) bool {
b.mu.Lock()
Comment thread
rdimitrov marked this conversation as resolved.
defer b.mu.Unlock()
if b.closed {
b.active--
return false
}
b.wg.Add(1)
go func() {
defer b.wg.Done()
defer b.releaseSlot()
review()
}()
return true
}

func (b *backgroundReviews) releaseSlot() {
b.mu.Lock()
b.active--
b.mu.Unlock()
}

// drain stops new reviews, cancels in-flight ones, and waits up to the grace for
// them to unwind. It is idempotent and reports whether every review unwound.
func (b *backgroundReviews) drain() bool {
b.mu.Lock()
b.closed = true
cancel := b.cancel
b.mu.Unlock()
if cancel == nil {
return true
}
cancel()
done := make(chan struct{})
go func() {
b.wg.Wait()
close(done)
}()
timer := time.NewTimer(backgroundReviewDrainGrace)
defer timer.Stop()
select {
case <-done:
return true
case <-timer.C:
return false
}
}

func (r *Run) drainBackgroundReviews(ctx context.Context) {
if !r.backgroundReviews.drain() {
r.diag.Log(ctx, port.LevelWarn, "abandoning yolo advisory guardrail reviews that did not stop within the run-end grace",
"grace", backgroundReviewDrainGrace.String())
}
}

func advisoryDisposition(result ToolReviewResult, err error, clean string) string {
if err != nil || result.Assessment != ReviewAcceptable {
return "pass_advisory"
}
return clean
}

// launchBackgroundReview runs assess off the dispatch path, records its checker
// usage, and publishes the advisory event built by publish unless run end
// cancelled the review first. release always runs once.
func (r *Run) launchBackgroundReview(ctx context.Context, assess func() session.AuxiliaryUsage, publish func() session.Event, release func()) {
if !r.backgroundReviews.start(func() {
defer release()
usage := assess()
r.enqueueAuxiliaryUsageWhileActive(ctx, RemapAuxiliaryUsage(ctx, r.diag, session.UsageKindGuardrail, usage))
if ctx.Err() == nil {
r.children.safeEmit(publish())
}
}) {
release()
}
}

// startBackgroundActionReview reviews an opted-in yolo action without delaying
// it. The call has already been authorized and proceeds regardless of verdict.
func (e *Engine) startBackgroundActionReview(r *Run, sess *session.Session, env tool.Environment, turnIdx int, call session.ToolCall) {
if r.reviewRoot == nil || !yoloBackgroundReview(r.reviewRoot.reviewer, sess.Mode, call.Name, ReviewJobAction) {
return
Comment thread
rdimitrov marked this conversation as resolved.
}
bg, ok := r.backgroundReviews.reserve(r.ctx)
if !ok {
return
}
sessID := sess.ID
Comment thread
rdimitrov marked this conversation as resolved.
a := e.prepareActionAssessment(bg, r, sess, env, call, true)
r.launchBackgroundReview(bg, func() session.AuxiliaryUsage {
assessAction(bg, r, &a)
return a.usage
}, func() session.Event {
result := a.result
if a.err != nil {
result.Assessment = ReviewUnresolved
}
r.publishActionDetail(bg, sessID, a.action.request.ReviewID, a.action.request, result, a.err)
guardrail := reviewMachinePayload(r.reviewRoot.reviewer, a.action.request, result, a.err, advisoryDisposition(result, a.err, "execute"))
return session.Event{Type: session.EvHook, Turn: turnIdx, Hook: &session.HookPayload{Phase: string(governance.PhasePreToolUse), Tool: call.Name, Decision: session.HookInfo, CallID: call.ID, Guardrail: guardrail}}
}, func() {
a.cancel()
if a.action.close != nil {
a.action.close()
}
})
}

// startBackgroundInboundReview inspects an opted-in yolo result that has
// already been released to the model; the verdict is advisory only.
func (e *Engine) startBackgroundInboundReview(r *Run, sess *session.Session, env tool.Environment, turnIdx int, call session.ToolCall, result session.ToolResult) {
if r.reviewRoot == nil || !yoloBackgroundReview(r.reviewRoot.reviewer, sess.Mode, call.Name, ReviewJobInbound) {
return
}
bg, ok := r.backgroundReviews.reserve(r.ctx)
if !ok {
return
}
sessID := sess.ID
a := inboundAssessment{applies: true, mode: sess.Mode, snapshot: true}
e.prepareInboundRequest(bg, r, sess, env, call, result, &a)
r.launchBackgroundReview(bg, func() session.AuxiliaryUsage {
assessInbound(bg, r, &a)
return a.usage
}, func() session.Event {
r.publishInboundDetail(bg, sessID, a)
return inboundReviewEvent(r, turnIdx, call, a, advisoryDisposition(a.result, a.err, "release_result"))
}, func() {
a.cancel()
if a.close != nil {
a.close()
}
})
}
11 changes: 8 additions & 3 deletions engine/agent/dispatch.go
Original file line number Diff line number Diff line change
Expand Up @@ -361,7 +361,7 @@ func (e *Engine) runReadBatch(ctx context.Context, r *Run, sess *session.Session

// Human permission and pre-hook waits are complete. Prepare each sibling
// serially, launching its private checker before preparing the next.
e.fanOutReadBatchActionAssessments(ctx, r, sess, env, prepared)
e.fanOutReadBatchActionAssessments(ctx, r, sess, env, turnIdx, prepared)
if ctx.Err() != nil {
r.recordCompletedActionUsage(ctx, sess, prepared)
return out, true
Expand Down Expand Up @@ -422,7 +422,7 @@ func (e *Engine) runReadBatch(ctx context.Context, r *Run, sess *session.Session

// fanOutReadBatchActionAssessments starts every eligible private action review,
// then joins them before the ordered decision drain begins.
func (e *Engine) fanOutReadBatchActionAssessments(ctx context.Context, r *Run, sess *session.Session, env tool.Environment, prepared []readBatchPending) {
func (e *Engine) fanOutReadBatchActionAssessments(ctx context.Context, r *Run, sess *session.Session, env tool.Environment, turnIdx int, prepared []readBatchPending) {
var reviewWG sync.WaitGroup
for i := range prepared {
if ctx.Err() != nil {
Expand All @@ -433,13 +433,15 @@ func (e *Engine) fanOutReadBatchActionAssessments(ctx context.Context, r *Run, s
continue
}
if applies, _ := reviewPolicy(r.reviewRoot.reviewer, sess.Mode, p.call.Name, ReviewJobAction, false); applies {
assessment := e.prepareActionAssessment(ctx, r, sess, env, p.call)
assessment := e.prepareActionAssessment(ctx, r, sess, env, p.call, false)
p.assessment = &assessment
reviewWG.Add(1)
go func(a *actionReviewAssessment) {
defer reviewWG.Done()
assessAction(ctx, r, a)
}(p.assessment)
} else {
e.startBackgroundActionReview(r, sess, env, turnIdx, p.call)
}
}
reviewWG.Wait()
Expand Down Expand Up @@ -1754,6 +1756,9 @@ func (e *Engine) executePrivate(ctx context.Context, r *Run, sess *session.Sessi
if r.reviewRoot != nil && r.reviewRoot.reviewer != nil {
assessment = e.prepareInboundAssessment(ctx, r, sess, env, c, res)
assessInbound(ctx, r, &assessment)
if !assessment.applies {
e.startBackgroundInboundReview(r, sess, env, turnIdx, c, res)
}
}
return executionRecord{
result: res, queued: queued, duration: dur, assessment: assessment, postEvents: postEvents,
Expand Down
Loading
Loading