fix(audit): wire subagent handoff observations end to end

The channel counting how often a delegated child closes the way it was asked
to reached nothing. No type implemented SubagentHandoffSink anywhere in the
tree, AuditForwarder had no method for it, and it was never registered in
capabilityContracts — so the type assertion inside RecordSubagentHandoff
matched nothing at the first wrapper and the audit was dropped. The producer
had a struct, an interface and a green unit test, which is exactly why nothing
noticed: conformance says a method exists, and only a stack says a value
arrives.

Registering it is the fix that matters. The registry is what forces every
wrapper and the shared forwarder to carry a channel, and the reason this one
escaped is that it never joined the closed set. AuditForwarder is embedded by
notify, telemetry, control, traystate, stats and the CLI sinks, so one method
there restores the channel for all of them.

The two guards that should have caught it kept their own hand-written lists of
capabilities, which pass for every capability they have not heard of. The
forwarder guard now reads capabilityContracts directly, and moves into
internal/event where that registry lives rather than exporting it for a test.
The cross-package wrapper guard still keeps a list; it is extended here, and
switching it to the registry is blocked on two channels that have no terminal
consumer at all (RunBudget, WorkspaceMutation).

Consumers follow the project-check probe's split: run-metrics aggregates,
trajectory keeps the whole observation. The metrics denominator is Judged —
expected to report and not killed by the provider — because a run that never
reached the point of closing is not a refusal, and counting it as one answers a
different question than the one being asked.

Nothing about delegation changes: no prompt byte, no handoff gate, no
completion semantics. This only makes the existing observation observable.
This commit is contained in:
esengine committed 2026-08-29 07:09:15 -07:00
1 parent e94617bb4e
commit 3f02f9fc7b
12 files changed
+274 -40

No files matched your search

+1 -20
View File
@@ -13,26 +13,6 @@ import (
"reasonix/internal/trajectory"
)
// The shared forwarder is why a new capability no longer has to be repeated at
// every layer, so it must itself be complete.
func TestAuditForwarderCoversEveryCapability(t *testing.T) {
fwd := reflect.TypeFor[event.AuditForwarder]()
for name, want := range map[string]reflect.Type{
"DelegationAuditSink": reflect.TypeFor[event.DelegationAuditSink](),
"ReadinessAuditSink": reflect.TypeFor[event.ReadinessAuditSink](),
"ContractShadowAuditSink": reflect.TypeFor[event.ContractShadowAuditSink](),
"CompletionReportAuditSink": reflect.TypeFor[event.CompletionReportAuditSink](),
"MemoryRecallSink": reflect.TypeFor[event.MemoryRecallSink](),
"OutcomeProgressSink": reflect.TypeFor[event.OutcomeProgressSink](),
"ProtocolRecoveryAuditSink": reflect.TypeFor[event.ProtocolRecoveryAuditSink](),
"TurnCompletionSink": reflect.TypeFor[event.TurnCompletionSink](),
} {
if !fwd.Implements(want) {
t.Errorf("event.AuditForwarder does not forward %s; every embedder loses it", name)
}
}
}
// A wrapper that forwards one audit but forgets another drops its data
// silently, with the tests still green. Pure forwarders should embed
// event.AuditForwarder instead of repeating the nine methods.
@@ -46,6 +26,7 @@ func TestAuditCapabilitiesAreForwardedByEveryWrapper(t *testing.T) {
"OutcomeProgressSink": reflect.TypeFor[event.OutcomeProgressSink](),
"ProtocolRecoveryAuditSink": reflect.TypeFor[event.ProtocolRecoveryAuditSink](),
"TurnCompletionSink": reflect.TypeFor[event.TurnCompletionSink](),
"SubagentHandoffSink": reflect.TypeFor[event.SubagentHandoffSink](),
}
wrappers := []event.Sink{
event.Sync(event.Discard),
+78 -19
View File
@@ -87,25 +87,40 @@ type RunMetrics struct {
// Delegation counters let one model be compared across orchestration arms
// without scraping prose. Child tool calls are already split out as
// SubagentToolCalls below; parent calls are ToolCalls minus that.
SubagentRuns int `json:"subagent_runs,omitempty"`
SubagentNestedRuns int `json:"subagent_nested_runs,omitempty"`
SubagentMutations int `json:"subagent_mutations,omitempty"`
CompletionReports int `json:"completion_reports,omitempty"`
CompletionsProsedOnly int `json:"completions_prose_only,omitempty"`
FalseCompletions int `json:"false_completions,omitempty"`
CriterionDowngrades int `json:"criterion_downgrades,omitempty"`
WriteScopeViolations int `json:"write_scope_violations,omitempty"`
DuplicateWorkPaths int `json:"duplicate_work_paths,omitempty"`
ParentScopeHints int `json:"parent_scope_hints,omitempty"`
ParentNamedFiles int `json:"parent_named_files,omitempty"`
ChildEvidencePaths int `json:"child_evidence_paths,omitempty"`
ChildDiscoveredPaths int `json:"child_discovered_paths,omitempty"`
MissingReasoningDetected int `json:"missing_reasoning_detected,omitempty"`
MissingReasoningRetries int `json:"missing_reasoning_retries,omitempty"`
MissingReasoningRecovered int `json:"missing_reasoning_recovered,omitempty"`
MissingReasoningReplaced int `json:"missing_reasoning_retry_replaced_response,omitempty"`
MissingReasoningSuppressed int `json:"missing_reasoning_retry_suppressed,omitempty"`
MissingReasoningFallbacks int `json:"missing_reasoning_fallbacks,omitempty"`
SubagentRuns int `json:"subagent_runs,omitempty"`
SubagentNestedRuns int `json:"subagent_nested_runs,omitempty"`
SubagentMutations int `json:"subagent_mutations,omitempty"`
CompletionReports int `json:"completion_reports,omitempty"`
CompletionsProsedOnly int `json:"completions_prose_only,omitempty"`
FalseCompletions int `json:"false_completions,omitempty"`
CriterionDowngrades int `json:"criterion_downgrades,omitempty"`
WriteScopeViolations int `json:"write_scope_violations,omitempty"`
DuplicateWorkPaths int `json:"duplicate_work_paths,omitempty"`
ParentScopeHints int `json:"parent_scope_hints,omitempty"`
ParentNamedFiles int `json:"parent_named_files,omitempty"`
ChildEvidencePaths int `json:"child_evidence_paths,omitempty"`
ChildDiscoveredPaths int `json:"child_discovered_paths,omitempty"`
// Subagent handoff: whether a delegated child closed the way it was asked
// to. Judged is the denominator — expected to report and not killed by the
// provider — because a run that never got to close is not a refusal.
SubagentHandoffs int `json:"subagent_handoffs"`
SubagentHandoffExpected int `json:"subagent_handoff_expected"`
SubagentHandoffJudged int `json:"subagent_handoff_judged"`
SubagentHandoffAttempted int `json:"subagent_handoff_attempted"`
SubagentHandoffAccepted int `json:"subagent_handoff_accepted"`
SubagentHandoffMalformed int `json:"subagent_handoff_malformed"`
SubagentHandoffNotTried int `json:"subagent_handoff_never_attempted"`
SubagentHandoffClosedWith int `json:"subagent_handoff_closed_with_report"`
SubagentHandoffToolsAfter int `json:"subagent_handoff_tool_calls_after_report"`
SubagentHandoffLowered int `json:"subagent_handoff_lowered_claims"`
SubagentHandoffReadOnly int `json:"subagent_handoff_read_only"`
SubagentHandoffExits map[string]int `json:"subagent_handoff_exits,omitempty"`
MissingReasoningDetected int `json:"missing_reasoning_detected,omitempty"`
MissingReasoningRetries int `json:"missing_reasoning_retries,omitempty"`
MissingReasoningRecovered int `json:"missing_reasoning_recovered,omitempty"`
MissingReasoningReplaced int `json:"missing_reasoning_retry_replaced_response,omitempty"`
MissingReasoningSuppressed int `json:"missing_reasoning_retry_suppressed,omitempty"`
MissingReasoningFallbacks int `json:"missing_reasoning_fallbacks,omitempty"`
// Fanout times what the delegation counters only count: what the run waited
// on its fan-outs, against what the same work costs one at a time.
Fanout *FanoutMetrics `json:"fanout,omitempty"`
@@ -502,6 +517,50 @@ func (s *metricsSink) RecordReadinessAudit(a evidence.ReadinessAudit) {
s.m.ReadinessMissingMutation += a.MissingMutation
}
func (s *metricsSink) RecordSubagentHandoff(a event.SubagentHandoffAudit) {
if s == nil {
return
}
defer event.RecordSubagentHandoff(s.inner, a)
s.mu.Lock()
defer s.mu.Unlock()
s.m.SubagentHandoffs++
if a.Exit != "" {
if s.m.SubagentHandoffExits == nil {
s.m.SubagentHandoffExits = map[string]int{}
}
s.m.SubagentHandoffExits[a.Exit]++
}
if a.Expected {
s.m.SubagentHandoffExpected++
}
// A run the provider killed never reached the point of closing, so it
// leaves the compliance denominator instead of counting as a refusal.
if !a.Expected || a.Exit != "completed" {
return
}
s.m.SubagentHandoffJudged++
if a.ReadOnly {
s.m.SubagentHandoffReadOnly++
}
if a.Attempts == 0 {
s.m.SubagentHandoffNotTried++
} else {
s.m.SubagentHandoffAttempted++
}
if a.Accepted > 0 {
s.m.SubagentHandoffAccepted++
}
if a.Malformed > 0 {
s.m.SubagentHandoffMalformed++
}
if a.ReportRound > 0 && a.ToolCallsAfterReport == 0 {
s.m.SubagentHandoffClosedWith++
}
s.m.SubagentHandoffToolsAfter += a.ToolCallsAfterReport
s.m.SubagentHandoffLowered += a.LoweredClaims
}
func (s *metricsSink) RecordProjectCheckProbe(p event.ProjectCheckProbe) {
if s == nil {
return
@@ -0,0 +1,45 @@
package cli
import (
"testing"
"reasonix/internal/event"
)
// The compliance denominator is the point of this test: a child the provider
// killed never reached the point of closing, so it must leave the denominator
// rather than count as a refusal. Routing is asserted too — the audit is sent
// through a wrapper, not called on the sink directly, because a method that
// exists and is never reached is the bug this channel already had once.
func TestMetricsSinkCountsSubagentHandoffCompliance(t *testing.T) {
s := &metricsSink{inner: event.Discard}
chain := event.Sync(s)
event.RecordSubagentHandoff(chain, event.SubagentHandoffAudit{
Expected: true, Exit: "completed", ReadOnly: true,
})
event.RecordSubagentHandoff(chain, event.SubagentHandoffAudit{
Expected: true, Exit: "cancelled",
})
event.RecordSubagentHandoff(chain, event.SubagentHandoffAudit{
Expected: true, Exit: "completed",
Attempts: 1, Accepted: 1, ReportRound: 2, FinalRound: 2, LoweredClaims: 3,
})
m := s.m
if m.SubagentHandoffs != 3 || m.SubagentHandoffExpected != 3 || m.SubagentHandoffJudged != 2 {
t.Fatalf("totals = %d/%d/%d, want 3 observed, 3 expected, 2 judged",
m.SubagentHandoffs, m.SubagentHandoffExpected, m.SubagentHandoffJudged)
}
if m.SubagentHandoffNotTried != 1 || m.SubagentHandoffAttempted != 1 || m.SubagentHandoffAccepted != 1 {
t.Fatalf("never-attempted/attempted/accepted = %d/%d/%d, want 1/1/1 — the cancelled run must not be either",
m.SubagentHandoffNotTried, m.SubagentHandoffAttempted, m.SubagentHandoffAccepted)
}
if m.SubagentHandoffClosedWith != 1 || m.SubagentHandoffLowered != 3 || m.SubagentHandoffReadOnly != 1 {
t.Fatalf("closed/lowered/read-only = %d/%d/%d, want 1/3/1",
m.SubagentHandoffClosedWith, m.SubagentHandoffLowered, m.SubagentHandoffReadOnly)
}
if m.SubagentHandoffExits["completed"] != 2 || m.SubagentHandoffExits["cancelled"] != 1 {
t.Fatalf("exits = %v, want 2 completed and 1 cancelled", m.SubagentHandoffExits)
}
}
+4
View File
@@ -53,6 +53,10 @@ func (f AuditForwarder) RecordProjectCheckProbe(p ProjectCheckProbe) {
RecordProjectCheckProbe(f.Inner, p)
}
func (f AuditForwarder) RecordSubagentHandoff(a SubagentHandoffAudit) {
RecordSubagentHandoff(f.Inner, a)
}
// DelegationAuditSink receives one receipt per completed sub-agent run.
type DelegationAuditSink interface {
RecordDelegationAudit(a evidence.DelegationAudit)
+7
View File
@@ -160,6 +160,13 @@ func (c *coalescer) RecordProjectCheckProbe(p ProjectCheckProbe) {
RecordProjectCheckProbe(c.inner, p)
}
func (c *coalescer) RecordSubagentHandoff(a SubagentHandoffAudit) {
c.mu.Lock()
c.enqueueFlushLocked()
c.drainAndUnlock()
RecordSubagentHandoff(c.inner, a)
}
func (c *coalescer) RecordTurnCompletion() {
c.mu.Lock()
c.enqueueFlushLocked()
+6
View File
@@ -51,6 +51,12 @@ func (f *FanOut) RecordProjectCheckProbe(p ProjectCheckProbe) {
}
}
func (f *FanOut) RecordSubagentHandoff(a SubagentHandoffAudit) {
for _, s := range f.sinks {
RecordSubagentHandoff(s, a)
}
}
func (f *FanOut) RecordTurnCompletion() {
for _, s := range f.sinks {
RecordTurnCompletion(s)
+1
View File
@@ -18,6 +18,7 @@ var capabilityContracts = map[string]reflect.Type{
"ProtocolRecovery": reflect.TypeFor[ProtocolRecoveryAuditSink](),
"ReadinessAudit": reflect.TypeFor[ReadinessAuditSink](),
"RunBudget": reflect.TypeFor[RunBudgetSink](),
"SubagentHandoff": reflect.TypeFor[SubagentHandoffSink](),
"TurnCompletion": reflect.TypeFor[TurnCompletionSink](),
"WorkspaceMutation": reflect.TypeFor[WorkspaceMutationSink](),
}
+19
View File
@@ -1,6 +1,7 @@
package event
import (
"reflect"
"testing"
"time"
@@ -49,6 +50,9 @@ func (s *capturingSink) RecordRunBudget(RunBudgetSample) {
func (s *capturingSink) RecordProjectCheckProbe(ProjectCheckProbe) {
s.recorded = append(s.recorded, "ProjectCheckProbe")
}
func (s *capturingSink) RecordSubagentHandoff(SubagentHandoffAudit) {
s.recorded = append(s.recorded, "SubagentHandoff")
}
// recordAll drives every optional capability once through s.
func recordAll(s Sink) {
@@ -63,6 +67,21 @@ func recordAll(s Sink) {
RecordWorkspaceMutation(s, WorkspaceMutation{})
RecordRunBudget(s, RunBudgetSample{})
RecordProjectCheckProbe(s, ProjectCheckProbe{})
RecordSubagentHandoff(s, SubagentHandoffAudit{})
}
// AuditForwarder is why a new capability does not have to be repeated at every
// layer, so it must itself be complete. It reads the registry rather than a
// list kept beside it: a hand-written list passes for every capability it has
// never heard of, which is how a channel comes to be registered and forwarded
// by nothing.
func TestAuditForwarderCoversEveryCapability(t *testing.T) {
fwd := reflect.TypeFor[AuditForwarder]()
for name, want := range capabilityContracts {
if !fwd.Implements(want) {
t.Errorf("AuditForwarder does not forward %s; every embedder loses it", name)
}
}
}
func TestCapturingSinkCoversEveryCapability(t *testing.T) {
+8
View File
@@ -54,6 +54,14 @@ func (s *syncSink) RecordProjectCheckProbe(p ProjectCheckProbe) {
}
}
func (s *syncSink) RecordSubagentHandoff(a SubagentHandoffAudit) {
s.mu.Lock()
defer s.mu.Unlock()
if hs, ok := s.inner.(SubagentHandoffSink); ok {
hs.RecordSubagentHandoff(a)
}
}
func (s *syncSink) RecordTurnCompletion() {
s.mu.Lock()
defer s.mu.Unlock()
+51
View File
@@ -0,0 +1,51 @@
package trajectory
import (
"path/filepath"
"testing"
"time"
"reasonix/internal/event"
"reasonix/internal/testenv"
)
// The bug this guards: SubagentHandoffAudit had a producer, a struct, an
// interface and a green unit test, and still reached nothing — the type
// assertion inside event.RecordSubagentHandoff matched no sink in the chain.
// Conformance proves a method exists; only a stack proves a value arrives.
func TestSubagentHandoffSurvivesTheSessionSinkStack(t *testing.T) {
path := filepath.Join(testenv.TempDir(t), "run.trajectory.jsonl")
r, err := New(event.Discard, path, nil)
if err != nil {
t.Fatalf("New: %v", err)
}
// boot's assembly order: Coalesce -> Sync -> CostQuote -> recorder.
chain := event.Coalesce(event.Sync(event.NewCostQuoteSink(r, nil)), time.Millisecond)
event.RecordSubagentHandoff(chain, event.SubagentHandoffAudit{
Entrance: "read_only_task", Depth: 1, ReadOnly: true, Expected: true,
Exit: "completed", Attempts: 2, Accepted: 1, Malformed: 1,
ReportRound: 3, FinalRound: 4, ToolCallsAfterReport: 2,
ClaimedStatus: "complete", AdjudicatedStatus: "partial",
LoweredClaims: 1, Criteria: 3, Evidence: 4, Unresolved: 1,
})
if err := r.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
recs := readRecords(t, path)
if len(recs) != 1 || recs[0].SubagentHandoff == nil {
t.Fatalf("handoff never reached the recorder: %+v", recs)
}
got := *recs[0].SubagentHandoff
want := SubagentHandoff{
Entrance: "read_only_task", Depth: 1, ReadOnly: true, Expected: true,
Exit: "completed", Attempts: 2, Accepted: 1, Malformed: 1,
ReportRound: 3, FinalRound: 4, ToolCallsAfterReport: 2,
ClaimedStatus: "complete", AdjudicatedStatus: "partial",
LoweredClaims: 1, Criteria: 3, Evidence: 4, Unresolved: 1,
}
if got != want {
t.Fatalf("handoff record = %+v, want %+v", got, want)
}
}
+47
View File
@@ -36,6 +36,7 @@ type Record struct {
OutcomeProgress *OutcomeProgress `json:"outcome_progress,omitempty"`
MemoryRecall *MemoryRecall `json:"memory_recall,omitempty"`
ProjectCheck *ProjectCheckProbe `json:"project_check_probe,omitempty"`
SubagentHandoff *SubagentHandoff `json:"subagent_handoff,omitempty"`
// Deltas counts the streamed increments merged into this record; absent
// means one. TS stays the first increment's — that is the observation
// time-to-first-token readers key on — and EndTS carries the last's.
@@ -109,6 +110,29 @@ type ContractShadowAudit struct {
ReadyToFinalize bool `json:"ready_to_finalize,omitempty"`
}
// SubagentHandoff mirrors event.SubagentHandoffAudit: how one delegated child
// closed. Every field travels, because which cut explains a refusal — entrance,
// depth, read-only, how long the run was — is what the aggregate cannot say.
type SubagentHandoff struct {
Entrance string `json:"entrance,omitempty"`
Depth int `json:"depth,omitempty"`
ReadOnly bool `json:"read_only,omitempty"`
Expected bool `json:"expected,omitempty"`
Exit string `json:"exit,omitempty"`
Attempts int `json:"attempts,omitempty"`
Accepted int `json:"accepted,omitempty"`
Malformed int `json:"malformed,omitempty"`
ReportRound int `json:"report_round,omitempty"`
FinalRound int `json:"final_round,omitempty"`
ToolCallsAfterReport int `json:"tool_calls_after_report,omitempty"`
ClaimedStatus string `json:"claimed_status,omitempty"`
AdjudicatedStatus string `json:"adjudicated_status,omitempty"`
LoweredClaims int `json:"lowered_claims,omitempty"`
Criteria int `json:"criteria,omitempty"`
Evidence int `json:"evidence,omitempty"`
Unresolved int `json:"unresolved,omitempty"`
}
// ProjectCheckProbe mirrors event.ProjectCheckProbe: the shadow comparison
// between the readiness gate's project-check derivation and the ledger's
// obligations. Identities are recorded, not counted — a class is only
@@ -316,6 +340,29 @@ func (r *Recorder) RecordReadinessAudit(a evidence.ReadinessAudit) {
event.RecordReadinessAudit(r.inner, a)
}
func (r *Recorder) RecordSubagentHandoff(a event.SubagentHandoffAudit) {
r.append(Record{SubagentHandoff: &SubagentHandoff{
Entrance: a.Entrance,
Depth: a.Depth,
ReadOnly: a.ReadOnly,
Expected: a.Expected,
Exit: a.Exit,
Attempts: a.Attempts,
Accepted: a.Accepted,
Malformed: a.Malformed,
ReportRound: a.ReportRound,
FinalRound: a.FinalRound,
ToolCallsAfterReport: a.ToolCallsAfterReport,
ClaimedStatus: a.ClaimedStatus,
AdjudicatedStatus: a.AdjudicatedStatus,
LoweredClaims: a.LoweredClaims,
Criteria: a.Criteria,
Evidence: a.Evidence,
Unresolved: a.Unresolved,
}})
event.RecordSubagentHandoff(r.inner, a)
}
func (r *Recorder) RecordProjectCheckProbe(p event.ProjectCheckProbe) {
rec := &ProjectCheckProbe{
Declared: p.Declared,
+7 -1
View File
@@ -44,7 +44,13 @@ func readRecords(t *testing.T, path string) []Record {
t.Fatalf("read trajectory: %v", err)
}
var out []Record
for line := range strings.SplitSeq(strings.TrimSpace(string(data)), "\n") {
// An empty file means nothing arrived, which is a real outcome to report as
// itself rather than as a malformed record.
trimmed := strings.TrimSpace(string(data))
if trimmed == "" {
return nil
}
for line := range strings.SplitSeq(trimmed, "\n") {
var r Record
if err := json.Unmarshal([]byte(line), &r); err != nil {
t.Fatalf("bad record %q: %v", line, err)