5d30b85 fix(subagents): pair live trace activity · zut

M packages/agent/subagent_worker.go +3 -1


@@ -249,7 +249,6 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
249	249		}
250	250		turnID := fmt.Sprintf("turn-%d", step)
251	251		em.setTurnID(turnID)
252		-	defer em.setTurnID("")
253	252		c, cancel := subagentTurnContext(ctx, args.SubagentTurnTimeout)
254	253		cancelFn = cancel
255	254		mu.Unlock()

@@ -356,6 +355,9 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
356	355
357	356		cancel()
358	357		mu.Lock()
	358	+	// Clear the emitter identity before reopening the busy gate: a newly
	359	+	// admitted turn must not race this completed turn's cleanup.
	360	+	em.setTurnID("")
359	361		busyTurn = false
360	362		cancelFn = nil
361	363		mu.Unlock()

 		}
 		turnID := fmt.Sprintf("turn-%d", step)
 		em.setTurnID(turnID)
-		defer em.setTurnID("")
 		c, cancel := subagentTurnContext(ctx, args.SubagentTurnTimeout)
 		cancelFn = cancel
 		mu.Unlock()
 		cancel()
 		mu.Lock()
+		// Clear the emitter identity before reopening the busy gate: a newly
+		// admitted turn must not race this completed turn's cleanup.
+		em.setTurnID("")
 		busyTurn = false
 		cancelFn = nil
 		mu.Unlock()

M packages/agent/subagents/projection.go +7 -3


@@ -193,12 +193,15 @@ func ProjectTrace(events []TraceEvent) map[string]AgentTraceView {
193	193		}
194	194
195	195		// ObservationFor returns the last observation only when it belongs to this
196		-	// operation's turn. A prior turn's stream is not live activity for a resumed
197		-	// or subsequent turn.
	196	+	// operation's turn and, when identified, its tool call. A prior turn's stream
	197	+	// or another tool's output is not live activity for this operation.
198	198		func (v AgentTraceView) ObservationFor(operation Operation) *LiveObservation {
199	199		if v.LastObservation == nil \|\| operation.TurnID == "" \|\| v.LastObservation.TurnID == "" \|\| operation.TurnID != v.LastObservation.TurnID {
200	200		return nil
201	201		}
	202	+	if v.LastObservation.CallID != "" && v.LastObservation.CallID != operation.CallID {
	203	+	return nil
	204	+	}
202	205		return v.LastObservation
203	206		}
204	207

@@ -222,9 +225,10 @@ func traceObservation(event TraceEvent) *LiveObservation {
222	225		source, _ = event.Data["worker_event"].(string)
223	226		}
224	227		name, _ := event.Data["name"].(string)
	228	+	callID, _ := event.Data["call_id"].(string)
225	229		switch typeName {
226	230		case "assistant.stream.observed", "reasoning.stream.observed", "tool.output.observed", "worker.protocol.observed":
227		-	return &LiveObservation{Type: typeName, Source: source, At: event.Timestamp, TurnID: event.TurnID, Name: name}
	231	+	return &LiveObservation{Type: typeName, Source: source, At: event.Timestamp, TurnID: event.TurnID, CallID: callID, Name: name}
228	232		}
229	233		return nil
230	234		}

 }
 // ObservationFor returns the last observation only when it belongs to this
-// operation's turn. A prior turn's stream is not live activity for a resumed
+// operation's turn and, when identified, its tool call. A prior turn's stream
-// or subsequent turn.
+// or another tool's output is not live activity for this operation.
 func (v AgentTraceView) ObservationFor(operation Operation) *LiveObservation {
 	if v.LastObservation == nil || operation.TurnID == "" || v.LastObservation.TurnID == "" || operation.TurnID != v.LastObservation.TurnID {
 		return nil
 	}
+	if v.LastObservation.CallID != "" && v.LastObservation.CallID != operation.CallID {
+		return nil
+	}
 	return v.LastObservation
 }
 		source, _ = event.Data["worker_event"].(string)
 	}
 	name, _ := event.Data["name"].(string)
+	callID, _ := event.Data["call_id"].(string)
 	switch typeName {
 	case "assistant.stream.observed", "reasoning.stream.observed", "tool.output.observed", "worker.protocol.observed":
-		return &LiveObservation{Type: typeName, Source: source, At: event.Timestamp, TurnID: event.TurnID, Name: name}
+		return &LiveObservation{Type: typeName, Source: source, At: event.Timestamp, TurnID: event.TurnID, CallID: callID, Name: name}
 	}
 	return nil
 }

M packages/agent/subagents/projection_test.go +19 -0


@@ -79,6 +79,25 @@ func TestProjectTraceSelectsSpecificActivityAndProjectsResultDelivery(t *testing
79	79		}
80	80		}
81	81
	82	+	func TestProjectTraceOmitsObservationForAnotherToolInSameTurn(t *testing.T) {
	83	+	started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
	84	+	view := ProjectTrace([]TraceEvent{
	85	+	{Timestamp: started, Type: "turn.started", AgentID: "agent-1", TurnID: "turn-1"},
	86	+	{Timestamp: started.Add(time.Second), Type: "tool.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "one", "name": "bash"}},
	87	+	{Timestamp: started.Add(2 * time.Second), Type: "tool.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "two", "name": "read"}},
	88	+	{Timestamp: started.Add(3 * time.Second), Type: "tool.output.observed", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "two", "name": "read"}},
	89	+	})["agent-1"]
	90	+	if view.PrimaryOperation == nil \|\| view.PrimaryOperation.CallID != "one" {
	91	+	t.Fatalf("primary operation = %#v, want call one", view.PrimaryOperation)
	92	+	}
	93	+	if observation := view.ObservationFor(*view.PrimaryOperation); observation != nil {
	94	+	t.Fatalf("another tool's observation = %#v, want omitted", observation)
	95	+	}
	96	+	if got, want := view.Summary(), "tool bash open"; got != want {
	97	+	t.Fatalf("summary = %q, want %q", got, want)
	98	+	}
	99	+	}
	100	+
82	101		func TestProjectTraceOmitsPriorTurnObservation(t *testing.T) {
83	102		started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
84	103		view := ProjectTrace([]TraceEvent{


@@ -79,6 +79,25 @@ func TestProjectTraceSelectsSpecificActivityAndProjectsResultDelivery(t *testing
79	}	79		}
80	}	80		}
81		81
		82	+	func TestProjectTraceOmitsObservationForAnotherToolInSameTurn(t *testing.T) {
		83	+	started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
		84	+	view := ProjectTrace([]TraceEvent{
		85	+	{Timestamp: started, Type: "turn.started", AgentID: "agent-1", TurnID: "turn-1"},
		86	+	{Timestamp: started.Add(time.Second), Type: "tool.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "one", "name": "bash"}},
		87	+	{Timestamp: started.Add(2 * time.Second), Type: "tool.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "two", "name": "read"}},
		88	+	{Timestamp: started.Add(3 * time.Second), Type: "tool.output.observed", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "two", "name": "read"}},
		89	+	})["agent-1"]
		90	+	if view.PrimaryOperation == nil \|\| view.PrimaryOperation.CallID != "one" {
		91	+	t.Fatalf("primary operation = %#v, want call one", view.PrimaryOperation)
		92	+	}
		93	+	if observation := view.ObservationFor(*view.PrimaryOperation); observation != nil {
		94	+	t.Fatalf("another tool's observation = %#v, want omitted", observation)
		95	+	}
		96	+	if got, want := view.Summary(), "tool bash open"; got != want {
		97	+	t.Fatalf("summary = %q, want %q", got, want)
		98	+	}
		99	+	}
		100	+
82	func TestProjectTraceOmitsPriorTurnObservation(t *testing.T) {	101		func TestProjectTraceOmitsPriorTurnObservation(t *testing.T) {
83	started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)	102		started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC)
84	view := ProjectTrace([]TraceEvent{	103		view := ProjectTrace([]TraceEvent{

M packages/agent/subagents/runner.go +3 -1


@@ -909,7 +909,9 @@ func recordWorkerTrace(agent *Agent, ev Event) {
909	909		switch ev.Type {
910	910		case "turn_start", "turn_end":
911	911		if nestedTurn {
912		-	if ev.Type == "turn_end" {
	912	+	if ev.Type == "turn_start" {
	913	+	traceType = "provider.request.started"
	914	+	} else {
913	915		traceType = "provider.request.finished"
914	916		}
915	917		} else if ev.Type == "turn_start" {

 	switch ev.Type {
 	case "turn_start", "turn_end":
 		if nestedTurn {
-			if ev.Type == "turn_end" {
+			if ev.Type == "turn_start" {
+				traceType = "provider.request.started"
+			} else {
 				traceType = "provider.request.finished"
 			}
 		} else if ev.Type == "turn_start" {

M packages/agent/subagents/runner_test.go +12 -8


@@ -807,6 +807,7 @@ func TestRecordWorkerTraceNormalizesSafeLiveObservations(t *testing.T) {
807	807		agent := &Agent{ID: "agent-1", trace: trace}
808	808		recordWorkerTrace(agent, Event{Type: EventMessageDelta, TurnID: "turn-1", Data: map[string]any{"delta": "sensitive answer"}})
809	809		recordWorkerTrace(agent, Event{Type: "request_started", TurnID: "turn-1", Data: map[string]any{"provider": "test", "model": "test"}})
	810	+	recordWorkerTrace(agent, Event{Type: "turn_start", TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})
810	811		recordWorkerTrace(agent, Event{Type: "turn_end", TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})
811	812		recordWorkerTrace(agent, Event{Type: "tool_call", TurnID: "turn-1", Data: map[string]any{"id": "call-1", "name": "bash", "args": "sensitive arguments"}})
812	813		recordWorkerTrace(agent, Event{Type: "future_event", TurnID: "turn-1", Data: map[string]any{"secret": "hidden"}})

@@ -814,7 +815,7 @@ func TestRecordWorkerTraceNormalizesSafeLiveObservations(t *testing.T) {
814	815		t.Fatal(err)
815	816		}
816	817		events := trace.Events()
817		-	if len(events) != 5 {
	818	+	if len(events) != 6 {
818	819		t.Fatalf("events = %#v", events)
819	820		}
820	821		if got, want := events[0].Type, "assistant.stream.observed"; got != want {

@@ -826,25 +827,28 @@ func TestRecordWorkerTraceNormalizesSafeLiveObservations(t *testing.T) {
826	827		if got, want := events[1].Type, "provider.request.started"; got != want {
827	828		t.Fatalf("type = %q, want %q", got, want)
828	829		}
829		-	if got, want := events[2].Type, "provider.request.finished"; got != want {
	830	+	if got, want := events[2].Type, "provider.request.started"; got != want {
830	831		t.Fatalf("type = %q, want %q", got, want)
831	832		}
832		-	if got, want := events[3].Type, "tool.started"; got != want {
	833	+	if got, want := events[3].Type, "provider.request.finished"; got != want {
833	834		t.Fatalf("type = %q, want %q", got, want)
834	835		}
835		-	if got, _ := events[3].Data["call_id"].(string); got != "call-1" {
	836	+	if got, want := events[4].Type, "tool.started"; got != want {
	837	+	t.Fatalf("type = %q, want %q", got, want)
	838	+	}
	839	+	if got, _ := events[4].Data["call_id"].(string); got != "call-1" {
836	840		t.Fatalf("tool call_id = %q", got)
837	841		}
838		-	if _, found := events[3].Data["args"]; found {
	842	+	if _, found := events[4].Data["args"]; found {
839	843		t.Fatalf("tool arguments leaked: %#v", events[2].Data)
840	844		}
841		-	if got, want := events[4].Type, "worker.protocol.observed"; got != want {
	845	+	if got, want := events[5].Type, "worker.protocol.observed"; got != want {
842	846		t.Fatalf("type = %q, want %q", got, want)
843	847		}
844		-	if got, _ := events[4].Data["source_event"].(string); got != "future.event" {
	848	+	if got, _ := events[5].Data["source_event"].(string); got != "future.event" {
845	849		t.Fatalf("source_event = %q", got)
846	850		}
847		-	if _, found := events[4].Data["secret"]; found {
	851	+	if _, found := events[5].Data["secret"]; found {
848	852		t.Fatalf("worker payload leaked: %#v", events[3].Data)
849	853		}
850	854		view := ProjectTrace(append([]TraceEvent{{Type: "turn.started", AgentID: "agent-1", TurnID: "turn-1"}}, events...))["agent-1"]

 	agent := &Agent{ID: "agent-1", trace: trace}
 	recordWorkerTrace(agent, Event{Type: EventMessageDelta, TurnID: "turn-1", Data: map[string]any{"delta": "sensitive answer"}})
 	recordWorkerTrace(agent, Event{Type: "request_started", TurnID: "turn-1", Data: map[string]any{"provider": "test", "model": "test"}})
+	recordWorkerTrace(agent, Event{Type: "turn_start", TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})
 	recordWorkerTrace(agent, Event{Type: "turn_end", TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})
 	recordWorkerTrace(agent, Event{Type: "tool_call", TurnID: "turn-1", Data: map[string]any{"id": "call-1", "name": "bash", "args": "sensitive arguments"}})
 	recordWorkerTrace(agent, Event{Type: "future_event", TurnID: "turn-1", Data: map[string]any{"secret": "hidden"}})
 		t.Fatal(err)
 	}
 	events := trace.Events()
-	if len(events) != 5 {
+	if len(events) != 6 {
 		t.Fatalf("events = %#v", events)
 	}
 	if got, want := events[0].Type, "assistant.stream.observed"; got != want {
 	if got, want := events[1].Type, "provider.request.started"; got != want {
 		t.Fatalf("type = %q, want %q", got, want)
 	}
-	if got, want := events[2].Type, "provider.request.finished"; got != want {
+	if got, want := events[2].Type, "provider.request.started"; got != want {
 		t.Fatalf("type = %q, want %q", got, want)
 	}
-	if got, want := events[3].Type, "tool.started"; got != want {
+	if got, want := events[3].Type, "provider.request.finished"; got != want {
 		t.Fatalf("type = %q, want %q", got, want)
 	}
-	if got, _ := events[3].Data["call_id"].(string); got != "call-1" {
+	if got, want := events[4].Type, "tool.started"; got != want {
+		t.Fatalf("type = %q, want %q", got, want)
+	}
+	if got, _ := events[4].Data["call_id"].(string); got != "call-1" {
 		t.Fatalf("tool call_id = %q", got)
 	}
-	if _, found := events[3].Data["args"]; found {
+	if _, found := events[4].Data["args"]; found {
 		t.Fatalf("tool arguments leaked: %#v", events[2].Data)
 	}
-	if got, want := events[4].Type, "worker.protocol.observed"; got != want {
+	if got, want := events[5].Type, "worker.protocol.observed"; got != want {
 		t.Fatalf("type = %q, want %q", got, want)
 	}
-	if got, _ := events[4].Data["source_event"].(string); got != "future.event" {
+	if got, _ := events[5].Data["source_event"].(string); got != "future.event" {
 		t.Fatalf("source_event = %q", got)
 	}
-	if _, found := events[4].Data["secret"]; found {
+	if _, found := events[5].Data["secret"]; found {
 		t.Fatalf("worker payload leaked: %#v", events[3].Data)
 	}
 	view := ProjectTrace(append([]TraceEvent{{Type: "turn.started", AgentID: "agent-1", TurnID: "turn-1"}}, events...))["agent-1"]