Merge pull request #8022 from esengine/fix/memorybench-recall-pipeline

fix(bench): make the MemoryBench pipeline actually observable

cmd/e2ebench/memorybench.go

@@ -33,6 +33,12 @@ func seedTaskMemory(taskDir, work string) ([]string, error) {
 	if err != nil {
 		return nil, err
 	}
+	// The child derives its project slug from Getwd, which returns the
+	// symlink-resolved path (/private/var vs /var on macOS); seed under the
+	// same identity or the store lands in a directory nobody reads.
+	if resolved, rErr := filepath.EvalSymlinks(absWork); rErr == nil {
+		absWork = resolved
+	}
 	pairs := [][2]string{
 		{filepath.Join(seeds, "project"), filepath.Join(stateHome, "projects", config.WorkspaceSlug(absWork), "memory")},
 		{filepath.Join(seeds, "global"), filepath.Join(stateHome, "memory", "global")},

internal/control/goalusage.go

@@ -84,6 +84,11 @@ func (t *goalUsageTee) RecordDelegationAdmission(a event.DelegationAdmissionAudi
 	event.RecordDelegationAdmission(t.inner, a)
 }
 
+// RecordMemoryRecall forwards the recall audit unchanged.
+func (t *goalUsageTee) RecordMemoryRecall(a event.MemoryRecallAudit) {
+	event.RecordMemoryRecall(t.inner, a)
+}
+
 // RecordContractShadow forwards the shadow contract audit unchanged.
 func (t *goalUsageTee) RecordContractShadow(a event.ContractShadowAudit) {
 	if t == nil || t.inner == nil {

internal/control/memory_recall_audit_test.go

@@ -0,0 +1,31 @@
+package control
+
+import (
+	"testing"
+
+	"reasonix/internal/agent"
+	"reasonix/internal/event"
+	"reasonix/internal/tool"
+)
+
+type recallAuditSink struct {
+	event.Sink
+	audits []event.MemoryRecallAudit
+}
+
+func (s *recallAuditSink) RecordMemoryRecall(a event.MemoryRecallAudit) {
+	s.audits = append(s.audits, a)
+}
+
+func TestComposeEmitsMemoryRecallAudit(t *testing.T) {
+	ag := agent.New(nil, tool.NewRegistry(), agent.NewSession("sys"), agent.Options{}, event.Discard)
+	sink := &recallAuditSink{Sink: event.Discard}
+	c := New(Options{Runner: ag, Executor: ag, Sink: sink})
+	_ = c.Compose("what is the canonical fast check command")
+	if len(sink.audits) != 1 {
+		t.Fatalf("audits = %d, want exactly one recall audit per composed user turn", len(sink.audits))
+	}
+	if sink.audits[0].Suppressed == "" && len(sink.audits[0].Hits) == 0 {
+		t.Fatalf("audit must explain itself: %+v", sink.audits[0])
+	}
+}

internal/event/coalesce.go

@@ -174,6 +174,13 @@ func (c *coalescer) RecordOutcomeProgress(sample evidence.OutcomeSample) {
 	RecordOutcomeProgress(c.inner, sample)
 }
 
+func (c *coalescer) RecordMemoryRecall(a MemoryRecallAudit) {
+	c.mu.Lock()
+	c.enqueueFlushLocked()
+	c.drainAndUnlock()
+	RecordMemoryRecall(c.inner, a)
+}
+
 func (c *coalescer) RecordDelegationAdmission(a DelegationAdmissionAudit) {
 	c.mu.Lock()
 	c.enqueueFlushLocked()

internal/event/sync.go

@@ -72,6 +72,14 @@ func (s *syncSink) RecordOutcomeProgress(sample evidence.OutcomeSample) {
 	}
 }
 
+func (s *syncSink) RecordMemoryRecall(a MemoryRecallAudit) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if mr, ok := s.inner.(MemoryRecallSink); ok {
+		mr.RecordMemoryRecall(a)
+	}
+}
+
 func (s *syncSink) RecordDelegationAdmission(a DelegationAdmissionAudit) {
 	s.mu.Lock()
 	defer s.mu.Unlock()

internal/notify/sink.go

@@ -61,6 +61,10 @@ func (s *Sink) RecordDelegationAdmission(a event.DelegationAdmissionAudit) {
 	event.RecordDelegationAdmission(s.inner, a)
 }
 
+func (s *Sink) RecordMemoryRecall(a event.MemoryRecallAudit) {
+	event.RecordMemoryRecall(s.inner, a)
+}
+
 // SendEvent applies the same notification rules for paths that do not emit through Sink.
 func SendEvent(sender Sender, cfg config.NotificationsConfig, e event.Event) {
 	if !cfg.Enabled || sender == nil {

internal/stats/recorder.go

@@ -184,6 +184,11 @@ func (r *Recorder) RecordOutcomeProgress(sample evidence.OutcomeSample) {
 	event.RecordOutcomeProgress(r.inner, sample)
 }
 
+// RecordMemoryRecall preserves the wrapped sink's audit capability.
+func (r *Recorder) RecordMemoryRecall(a event.MemoryRecallAudit) {
+	event.RecordMemoryRecall(r.inner, a)
+}
+
 // RecordDelegationAdmission preserves the wrapped sink's audit capability.
 func (r *Recorder) RecordDelegationAdmission(a event.DelegationAdmissionAudit) {
 	event.RecordDelegationAdmission(r.inner, a)

internal/telemetry/sink.go

@@ -143,6 +143,10 @@ func (s *sink) RecordDelegationAdmission(a event.DelegationAdmissionAudit) {
 	event.RecordDelegationAdmission(s.inner, a)
 }
 
+func (s *sink) RecordMemoryRecall(a event.MemoryRecallAudit) {
+	event.RecordMemoryRecall(s.inner, a)
+}
+
 func (s *sink) RecordProtocolRecovery(a event.ProtocolRecoveryAudit) {
 	add(s.counts, "tool_call_reasoning_recovery", string(a.Kind), 1)
 	event.RecordProtocolRecovery(s.inner, a)