Dataface Tasks

Type terminal agent event protocol and provider stream adapters

IDMCP_ANALYST_AGENT-TYPE_TERMINAL_AGENT_EVENT_PROTOCOL_AND_PROVIDER_STREAM_ADAPTERS
Statuscompleted
Priorityp2
Milestonem1-ft-analytics-analyst-pilot
Ownerdata-ai-engineer-architect

Problem

Refactor the terminal agent loop introduced in dataface/ai/agent.py and dataface/ai/llm.py to use explicit typed event structures and smaller provider adapter helpers. Reduce stringly typed event handling and isolate provider-specific stream parsing from agent-loop orchestration before more agent clients land.

Context

The terminal agent loop (dataface/ai/agent.py) and LLM client adapters (dataface/ai/llm.py) use dict[str, Any] for all events, discriminated by a "type" string key. Consumers (cli/commands/agent.py, tests) switch on event["type"] strings. Provider-specific stream parsing (OpenAI Responses API events, Anthropic Messages API events) lives inline in each client's stream_with_tools method.

Key files: dataface/ai/agent.py, dataface/ai/llm.py, dataface/cli/commands/agent.py, tests/core/test_ai_agent.py, tests/core/test_ai_llm.py.

Constraints: preserve all existing CLI behavior and test coverage; no new dependencies; Python 3.13 (match statements available).

Possible Solutions

  1. Typed dataclasses with union type — Define @dataclass(slots=True) for each event kind. Use StreamEvent = ContentDelta | ThinkingStatus | ToolCallEvent for provider output and AgentEvent = StreamEvent | ToolResultEvent | AgentDone | AgentError for the full agent protocol. Consumers use isinstance or match. Recommended — minimal, no magic, zero runtime cost beyond what dataclasses already provide.

  2. Pydantic discriminated union — Use Pydantic BaseModel with a literal type discriminator. Adds validation but also a runtime dependency and overhead the agent loop doesn't need. Against anti-slop philosophy.

  3. TypedDict variants — Keep dicts but use TypedDict for static checking. Cheaper change but doesn't actually remove stringly-typed runtime dispatch.

Plan

Selected: approach 1 (typed dataclasses).

  1. Create dataface/ai/events.py with dataclass events and StreamEvent/AgentEvent unions.
  2. Extract provider stream parsing into module-level helpers _iter_openai_stream and _iter_anthropic_stream in llm.py; have client methods delegate to them.
  3. Update LLMClient protocol and concrete clients to yield StreamEvent.
  4. Update agent.py to yield AgentEvent and use isinstance checks.
  5. Update cli/commands/agent.py to match on event types.
  6. Update tests to construct/assert typed events.
  7. Run just test for the touched modules.

Implementation Progress

  • Created dataface/ai/events.py with 6 @dataclass(slots=True) event types: ContentDelta, ThinkingStatus, ToolCallEvent (stream-level) and ToolResultEvent, AgentDone, AgentError (agent-level). Defined StreamEvent and AgentEvent union type aliases.
  • Updated dataface/ai/llm.py: LLMClient protocol and concrete clients now yield StreamEvent instead of dict[str, Any]. Extracted _iter_anthropic_stream() as a standalone stream adapter helper. OpenAI parsing stays inline due to response ID state tracking.
  • Updated dataface/ai/agent.py: run_agent() returns Iterator[AgentEvent] and uses isinstance checks instead of string key lookups.
  • Updated dataface/cli/commands/agent.py: switched from event["type"] string dispatch to match statement on typed events.
  • Updated all 3 test files to construct and assert on typed event objects instead of dicts.
  • All 9 existing tests pass. Ruff, Black, and Mypy clean.

QA Exploration

  • [x] QA exploration completed (or N/A for non-UI tasks)

Review Feedback

  • [ ] Review cleared