3cb020e fix(subagents): retain worker activity turn IDs · zut

M packages/agent/subagent_worker.go +14 -0


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

@@ -487,6 +489,7 @@ type subagentEmitter struct {
487	489		mu sync.Mutex
488	490		w *os.File
489	491		agentID string
	492	+	turnID string
490	493		}
491	494
492	495		func newSubagentEmitter(w os.File) subagentEmitter {

@@ -499,6 +502,14 @@ func (e *subagentEmitter) setProtocolIdentity(agentID string) {
499	502		e.mu.Unlock()
500	503		}
501	504
	505	+	// setTurnID associates subsequent worker events with the currently admitted
	506	+	// delegated turn. Core events do not carry the worker protocol turn ID.
	507	+	func (e *subagentEmitter) setTurnID(turnID string) {
	508	+	e.mu.Lock()
	509	+	e.turnID = turnID
	510	+	e.mu.Unlock()
	511	+	}
	512	+
502	513		func (e *subagentEmitter) emit(typ string, data map[string]any) {
503	514		if data == nil {
504	515		data = map[string]any{}

@@ -511,6 +522,9 @@ func (e *subagentEmitter) emit(typ string, data map[string]any) {
511	522
512	523		e.mu.Lock()
513	524		defer e.mu.Unlock()
	525	+	if turnID == "" {
	526	+	turnID = e.turnID
	527	+	}
514	528
515	529		wireType := canonicalWorkerEvent(typ)
516	530		envelope := subagents.NewEventEnvelope(wireType, e.agentID, turnID, payload)


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

@@ -487,6 +489,7 @@ type subagentEmitter struct {
487	mu sync.Mutex	489		mu sync.Mutex
488	w *os.File	490		w *os.File
489	agentID string	491		agentID string
		492	+	turnID string
490	}	493		}
491		494
492	func newSubagentEmitter(w os.File) subagentEmitter {	495		func newSubagentEmitter(w os.File) subagentEmitter {

@@ -499,6 +502,14 @@ func (e *subagentEmitter) setProtocolIdentity(agentID string) {
499	e.mu.Unlock()	502		e.mu.Unlock()
500	}	503		}
501		504
		505	+	// setTurnID associates subsequent worker events with the currently admitted
		506	+	// delegated turn. Core events do not carry the worker protocol turn ID.
		507	+	func (e *subagentEmitter) setTurnID(turnID string) {
		508	+	e.mu.Lock()
		509	+	e.turnID = turnID
		510	+	e.mu.Unlock()
		511	+	}
		512	+
502	func (e *subagentEmitter) emit(typ string, data map[string]any) {	513		func (e *subagentEmitter) emit(typ string, data map[string]any) {
503	if data == nil {	514		if data == nil {
504	data = map[string]any{}	515		data = map[string]any{}

@@ -511,6 +522,9 @@ func (e *subagentEmitter) emit(typ string, data map[string]any) {
511		522
512	e.mu.Lock()	523		e.mu.Lock()
513	defer e.mu.Unlock()	524		defer e.mu.Unlock()
		525	+	if turnID == "" {
		526	+	turnID = e.turnID
		527	+	}
514		528
515	wireType := canonicalWorkerEvent(typ)	529		wireType := canonicalWorkerEvent(typ)
516	envelope := subagents.NewEventEnvelope(wireType, e.agentID, turnID, payload)	530		envelope := subagents.NewEventEnvelope(wireType, e.agentID, turnID, payload)

M packages/agent/subagent_worker_test.go +8 -4


@@ -819,7 +819,8 @@ func TestSupervisorEmitterStdoutShapeMatchesSupervisorParser(t *testing.T) {
819	819		defer r.Close()
820	820
821	821		em := newSubagentEmitter(w)
822		-	em.emit("turn_start", map[string]any{"step": 1})
	822	+	em.setTurnID("turn-1")
	823	+	em.emit("text_delta", map[string]any{"delta": "sensitive partial answer"})
823	824		_ = w.Close()
824	825
825	826		body, err := io.ReadAll(r)

@@ -835,15 +836,18 @@ func TestSupervisorEmitterStdoutShapeMatchesSupervisorParser(t *testing.T) {
835	836		if err := json.Unmarshal(lines[0], &object); err != nil {
836	837		t.Fatalf("not valid json: %v\n%s", err, lines[0])
837	838		}
838		-	if object["type"] != "turn.started" {
	839	+	if object["type"] != "message.delta" {
839	840		t.Errorf("type field missing or wrong: %v", object["type"])
840	841		}
	842	+	if object["turn_id"] != "turn-1" {
	843	+	t.Errorf("turn_id = %v, want turn-1", object["turn_id"])
	844	+	}
841	845		if _, ok := object["timestamp"].(string); !ok {
842	846		t.Errorf("timestamp field missing: %v", object["timestamp"])
843	847		}
844	848		payload, ok := object["payload"].(map[string]any)
845		-	if !ok \|\| payload["step"] != float64(1) {
846		-	t.Errorf("payload step field missing or wrong: %v", object["payload"])
	849	+	if !ok \|\| payload["delta"] != "sensitive partial answer" {
	850	+	t.Errorf("payload delta field missing or wrong: %v", object["payload"])
847	851		}
848	852		}
849	853

 	defer r.Close()
 	em := newSubagentEmitter(w)
-	em.emit("turn_start", map[string]any{"step": 1})
+	em.setTurnID("turn-1")
+	em.emit("text_delta", map[string]any{"delta": "sensitive partial answer"})
 	_ = w.Close()
 	body, err := io.ReadAll(r)
 	if err := json.Unmarshal(lines[0], &object); err != nil {
 		t.Fatalf("not valid json: %v\n%s", err, lines[0])
 	}
-	if object["type"] != "turn.started" {
+	if object["type"] != "message.delta" {
 		t.Errorf("type field missing or wrong: %v", object["type"])
 	}
+	if object["turn_id"] != "turn-1" {
+		t.Errorf("turn_id = %v, want turn-1", object["turn_id"])
+	}
 	if _, ok := object["timestamp"].(string); !ok {
 		t.Errorf("timestamp field missing: %v", object["timestamp"])
 	}
 	payload, ok := object["payload"].(map[string]any)
-	if !ok || payload["step"] != float64(1) {
+	if !ok || payload["delta"] != "sensitive partial answer" {
-		t.Errorf("payload step field missing or wrong: %v", object["payload"])
+		t.Errorf("payload delta field missing or wrong: %v", object["payload"])
 	}
 }

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


@@ -905,13 +905,26 @@ func recordWorkerTrace(agent *Agent, ev Event) {
905	905		}
906	906		typeName := strings.ReplaceAll(ev.Type, "_", ".")
907	907		traceType := "worker.protocol.observed"
	908	+	nestedTurn, _ := ev.Data["nested_turn"].(bool)
908	909		switch ev.Type {
909		-	case "turn_start", EventTurnStarted:
	910	+	case "turn_start", "turn_end":
	911	+	if nestedTurn {
	912	+	if ev.Type == "turn_end" {
	913	+	traceType = "provider.request.finished"
	914	+	}
	915	+	} else if ev.Type == "turn_start" {
	916	+	traceType = "turn.started"
	917	+	} else {
	918	+	traceType = "turn.finished"
	919	+	}
	920	+	case EventTurnStarted:
910	921		traceType = "turn.started"
911		-	case "turn_end", EventTurnResult:
	922	+	case EventTurnResult:
912	923		traceType = "turn.finished"
913	924		case "turn_failed", EventTurnFailed:
914	925		traceType = "turn.failed"
	926	+	case "request_started":
	927	+	traceType = "provider.request.started"
915	928		case "tool_call", EventToolStarted:
916	929		traceType = "tool.started"
917	930		case "tool_result", EventToolFinished:

@@ -930,7 +943,11 @@ func recordWorkerTrace(agent *Agent, ev Event) {
930	943		if status, ok := ev.Data["status"].(string); ok && status != "" {
931	944		data["status"] = status
932	945		}
933		-	if callID, ok := ev.Data["call_id"].(string); ok && callID != "" {
	946	+	callID, _ := ev.Data["call_id"].(string)
	947	+	if callID == "" {
	948	+	callID, _ = ev.Data["id"].(string)
	949	+	}
	950	+	if callID != "" {
934	951		data["call_id"] = callID
935	952		}
936	953		agent.recordTrace(TraceEvent{Type: traceType, Timestamp: ev.Time, TurnID: ev.TurnID, Data: data})

 	}
 	typeName := strings.ReplaceAll(ev.Type, "_", ".")
 	traceType := "worker.protocol.observed"
+	nestedTurn, _ := ev.Data["nested_turn"].(bool)
 	switch ev.Type {
-	case "turn_start", EventTurnStarted:
+	case "turn_start", "turn_end":
+		if nestedTurn {
+			if ev.Type == "turn_end" {
+				traceType = "provider.request.finished"
+			}
+		} else if ev.Type == "turn_start" {
+			traceType = "turn.started"
+		} else {
+			traceType = "turn.finished"
+		}
+	case EventTurnStarted:
 		traceType = "turn.started"
-	case "turn_end", EventTurnResult:
+	case EventTurnResult:
 		traceType = "turn.finished"
 	case "turn_failed", EventTurnFailed:
 		traceType = "turn.failed"
+	case "request_started":
+		traceType = "provider.request.started"
 	case "tool_call", EventToolStarted:
 		traceType = "tool.started"
 	case "tool_result", EventToolFinished:
 	if status, ok := ev.Data["status"].(string); ok && status != "" {
 		data["status"] = status
 	}
-	if callID, ok := ev.Data["call_id"].(string); ok && callID != "" {
+	callID, _ := ev.Data["call_id"].(string)
+	if callID == "" {
+		callID, _ = ev.Data["id"].(string)
+	}
+	if callID != "" {
 		data["call_id"] = callID
 	}
 	agent.recordTrace(TraceEvent{Type: traceType, Timestamp: ev.Time, TurnID: ev.TurnID, Data: data})

M packages/agent/subagents/runner_test.go +27 -5


@@ -806,12 +806,15 @@ func TestRecordWorkerTraceNormalizesSafeLiveObservations(t *testing.T) {
806	806		t.Cleanup(func() { _ = trace.Close() })
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	+	recordWorkerTrace(agent, Event{Type: "request_started", TurnID: "turn-1", Data: map[string]any{"provider": "test", "model": "test"}})
	810	+	recordWorkerTrace(agent, Event{Type: "turn_end", TurnID: "turn-1", Data: map[string]any{"nested_turn": true}})
	811	+	recordWorkerTrace(agent, Event{Type: "tool_call", TurnID: "turn-1", Data: map[string]any{"id": "call-1", "name": "bash", "args": "sensitive arguments"}})
809	812		recordWorkerTrace(agent, Event{Type: "future_event", TurnID: "turn-1", Data: map[string]any{"secret": "hidden"}})
810	813		if err := trace.Flush(); err != nil {
811	814		t.Fatal(err)
812	815		}
813	816		events := trace.Events()
814		-	if len(events) != 2 {
	817	+	if len(events) != 5 {
815	818		t.Fatalf("events = %#v", events)
816	819		}
817	820		if got, want := events[0].Type, "assistant.stream.observed"; got != want {

@@ -820,14 +823,33 @@ func TestRecordWorkerTraceNormalizesSafeLiveObservations(t *testing.T) {
820	823		if _, found := events[0].Data["delta"]; found {
821	824		t.Fatalf("stream payload leaked: %#v", events[0].Data)
822	825		}
823		-	if got, want := events[1].Type, "worker.protocol.observed"; got != want {
	826	+	if got, want := events[1].Type, "provider.request.started"; got != want {
824	827		t.Fatalf("type = %q, want %q", got, want)
825	828		}
826		-	if got, _ := events[1].Data["source_event"].(string); got != "future.event" {
	829	+	if got, want := events[2].Type, "provider.request.finished"; got != want {
	830	+	t.Fatalf("type = %q, want %q", got, want)
	831	+	}
	832	+	if got, want := events[3].Type, "tool.started"; got != want {
	833	+	t.Fatalf("type = %q, want %q", got, want)
	834	+	}
	835	+	if got, _ := events[3].Data["call_id"].(string); got != "call-1" {
	836	+	t.Fatalf("tool call_id = %q", got)
	837	+	}
	838	+	if _, found := events[3].Data["args"]; found {
	839	+	t.Fatalf("tool arguments leaked: %#v", events[2].Data)
	840	+	}
	841	+	if got, want := events[4].Type, "worker.protocol.observed"; got != want {
	842	+	t.Fatalf("type = %q, want %q", got, want)
	843	+	}
	844	+	if got, _ := events[4].Data["source_event"].(string); got != "future.event" {
827	845		t.Fatalf("source_event = %q", got)
828	846		}
829		-	if _, found := events[1].Data["secret"]; found {
830		-	t.Fatalf("worker payload leaked: %#v", events[1].Data)
	847	+	if _, found := events[4].Data["secret"]; found {
	848	+	t.Fatalf("worker payload leaked: %#v", events[3].Data)
	849	+	}
	850	+	view := ProjectTrace(append([]TraceEvent{{Type: "turn.started", AgentID: "agent-1", TurnID: "turn-1"}}, events...))["agent-1"]
	851	+	if view.PrimaryOperation == nil \|\| view.PrimaryOperation.Type != "tool.started" \|\| view.PrimaryOperation.Name != "bash" {
	852	+	t.Fatalf("current-turn tool activity was not projected: %#v", view)
831	853		}
832	854		}
833	855

 	t.Cleanup(func() { _ = trace.Close() })
 	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_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"}})
 	if err := trace.Flush(); err != nil {
 		t.Fatal(err)
 	}
 	events := trace.Events()
-	if len(events) != 2 {
+	if len(events) != 5 {
 		t.Fatalf("events = %#v", events)
 	}
 	if got, want := events[0].Type, "assistant.stream.observed"; got != want {
 	if _, found := events[0].Data["delta"]; found {
 		t.Fatalf("stream payload leaked: %#v", events[0].Data)
 	}
-	if got, want := events[1].Type, "worker.protocol.observed"; got != want {
+	if got, want := events[1].Type, "provider.request.started"; got != want {
 		t.Fatalf("type = %q, want %q", got, want)
 	}
-	if got, _ := events[1].Data["source_event"].(string); got != "future.event" {
+	if got, want := events[2].Type, "provider.request.finished"; got != want {
+		t.Fatalf("type = %q, want %q", got, want)
+	}
+	if got, want := events[3].Type, "tool.started"; got != want {
+		t.Fatalf("type = %q, want %q", got, want)
+	}
+	if got, _ := events[3].Data["call_id"].(string); got != "call-1" {
+		t.Fatalf("tool call_id = %q", got)
+	}
+	if _, found := events[3].Data["args"]; found {
+		t.Fatalf("tool arguments leaked: %#v", events[2].Data)
+	}
+	if got, want := events[4].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" {
 		t.Fatalf("source_event = %q", got)
 	}
-	if _, found := events[1].Data["secret"]; found {
+	if _, found := events[4].Data["secret"]; found {
-		t.Fatalf("worker payload leaked: %#v", events[1].Data)
+		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"]
+	if view.PrimaryOperation == nil || view.PrimaryOperation.Type != "tool.started" || view.PrimaryOperation.Name != "bash" {
+		t.Fatalf("current-turn tool activity was not projected: %#v", view)
 	}
 }