Skip to content

Add agent attribution to verbose logs for parallel and for-each execution #16

@e-s-gh

Description

@e-s-gh

Problem

When Conductor runs parallel or for-each agent groups, verbose log output from all concurrent agents is interleaved with no identifying information. Tool calls, reasoning steps, and processing indicators are indistinguishable:

    ├─ 🔧 view
    │     args: {'path': '/some/file.py'}
    │  ⏳ Processing...
    ├─ 🔧 grep
    │     args: {'pattern': 'some_function', ...}
    │  ⏳ Processing...
    ├─ 🔧 view
    │     args: {'path': '/another/file.py'}

When 3+ agents run concurrently, it's impossible to determine which agent made which tool call. This makes debugging parallel workflows impractical and log files ungrepable for per-agent activity.

The problem is especially acute with for_each groups where every item shares the same agent definition name.

Proposed Solution

1. Pass agent name through to verbose logging (copilot.py)

Add an optional agent_name parameter to _send_and_wait and _log_event_verbose. When present, prefix all verbose output with [agent_name]:

# _send_and_wait signature
async def _send_and_wait(
    self, session, prompt, verbose_enabled, full_enabled,
    interrupt_signal=None, event_callback=None,
    agent_name: str | None = None,      # NEW
) -> SDKResponse:

# _log_event_verbose signature
def _log_event_verbose(
    self, event_type, event, full_mode,
    agent_name: str | None = None,      # NEW
) -> None:

Inside _log_event_verbose, every event type that builds Rich Text() output conditionally prepends the agent tag:

if agent_name:
    text.append(f"[{agent_name}] ", style="magenta")

Applies to: tool.execution_start, tool.execution_complete, assistant.reasoning, subagent.started, subagent.completed, assistant.turn_start

All call sites of _send_and_wait pass agent_name=agent.name.

2. Qualify for-each agent names with item keys (workflow.py)

In _execute_for_each_groupexecute_single_item, create a shallow copy of the agent with a qualified name before execution:

qualified_agent = for_each_group.agent.model_copy(
    update={"name": f"{for_each_group.agent.name}[{key}]"}
)
output = await executor.execute(qualified_agent, agent_context, ...)

This uses Pydantic's model_copy() — only the name field changes. The key comes from the for-each item (index or key_by value).

Static parallel groups don't need this change — each agent already has a unique name.

Expected Result

    ├─ [analyzer[item_a]] 🔧 view
    │     args: {'path': '/some/file.py'}
    │  [analyzer[item_a]] ⏳ Processing...
    ├─ [analyzer[item_b]] 🔧 grep
    │     args: {'pattern': 'some_function', ...}
    │  [analyzer[item_b]] 💭 Reasoning about the results...
    ├─ [analyzer[item_c]] 🔧 view
    │     args: {'path': '/another/file.py'}

Log files become greppable per-agent:

# Filter to one specific agent's activity
grep "\[analyzer\[item_a\]\]" log-file

# Count tool calls per agent
grep -c "\[analyzer\[.*\]\] 🔧" log-file

Scope

File Change
src/conductor/providers/copilot.py Add agent_name param to _send_and_wait and _log_event_verbose; prefix verbose events
src/conductor/engine/workflow.py For-each items use model_copy() to create qualified agent name

Backward Compatibility

Fully backward compatible:

  • agent_name defaults to None — no tag emitted for sequential agents
  • model_copy() creates a new object; original agent definition is untouched
  • No schema changes, no YAML syntax changes, no CLI changes
  • Existing log parsing patterns still work — the tag is prepended, not replacing content

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions