social-app/docs/plans/2026-03-11-agentscope-agent-route-migration.md

# AgentScope Agent Route Migration Implementation Plan

> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

**Goal:** Keep `/api/v1/agent/*` routes stable while fully replacing old `core/agent` runtime with `core/agentscope` runtime, AG-UI event pipeline, Redis streaming, and session/message persistence.

**Architecture:** Route handlers remain under `v1/agent`, but all runtime behavior moves to `core/agentscope` across five modules (`runtime`, `prompts`, `schemas`, `tools`, `events`). The `events` module owns AG-UI conversion, persistence, and Redis stream publishing/reading. Runtime orchestrator emits internal events only, then delegates to `events.pipeline` for normalization, persistence, and transport.

**Tech Stack:** FastAPI, SQLAlchemy async, Redis streams, Taskiq, AgentScope ReActAgent, LiteLLM proxy, Pydantic v2, pytest.

---

### Task 1: Define AgentScope Runtime Schemas

**Files:**
- Modify: `backend/src/core/agentscope/schemas/__init__.py`
- Create: `backend/src/core/agentscope/schemas/agent_runtime.py`
- Test: `backend/tests/unit/core/agentscope/schemas/test_agent_runtime_schemas.py`

**Step 1: Write failing schema tests**

```python
def test_run_command_schema_roundtrip() -> None:
    payload = {"threadId": "...", "runId": "...", "messages": []}
    model = RunCommand.model_validate(payload)
    assert model.model_dump(by_alias=True)["threadId"] == payload["threadId"]
```

**Step 2: Run tests to verify failure**

Run: `uv run pytest tests/unit/core/agentscope/schemas/test_agent_runtime_schemas.py -q`
Expected: FAIL because schema module/classes are missing.

**Step 3: Implement schemas**

```python
class RunCommand(BaseModel):
    thread_id: str = Field(alias="threadId")
    run_id: str = Field(alias="runId")
```

Also define: ResumeCommand, InternalRuntimeEvent, AgUiWireEvent, HistorySnapshotResponse, AcceptedTaskResponse.

**Step 4: Re-run tests**

Run: `uv run pytest tests/unit/core/agentscope/schemas/test_agent_runtime_schemas.py -q`
Expected: PASS.

**Step 5: Commit**

```bash
git add backend/src/core/agentscope/schemas/agent_runtime.py backend/src/core/agentscope/schemas/__init__.py backend/tests/unit/core/agentscope/schemas/test_agent_runtime_schemas.py
git commit -m "feat: add agentscope runtime schemas for agent routes"
```

### Task 2: Build Events Module (AG-UI + Redis + Persistence)

**Files:**
- Create: `backend/src/core/agentscope/events/pipeline.py`
- Create: `backend/src/core/agentscope/events/agui_codec.py`
- Create: `backend/src/core/agentscope/events/redis_bus.py`
- Create: `backend/src/core/agentscope/events/sse.py`
- Create: `backend/src/core/agentscope/events/store.py`
- Create: `backend/src/core/agentscope/events/__init__.py`
- Test: `backend/tests/unit/core/agentscope/events/test_agui_codec.py`
- Test: `backend/tests/unit/core/agentscope/events/test_sse.py`
- Test: `backend/tests/unit/core/agentscope/events/test_pipeline.py`

**Step 1: Write failing tests for codec/sse/pipeline**

```python
def test_codec_maps_internal_text_delta_to_agui() -> None:
    event = to_agui_wire(...)
    assert event["type"] == "TEXT_MESSAGE_CONTENT"
```

**Step 2: Run tests to verify failure**

Run: `uv run pytest tests/unit/core/agentscope/events -q`
Expected: FAIL due to missing modules.

**Step 3: Implement module**

```python
class AgentScopeEventPipeline:
    async def emit(self, event: InternalRuntimeEvent) -> str:
        wire = to_agui_wire(event)
        await self._store.persist(wire)
        return await self._redis.append(wire)
```

Implement SSE encoder and Redis read with cursor support.

**Step 4: Re-run tests**

Run: `uv run pytest tests/unit/core/agentscope/events -q`
Expected: PASS.

**Step 5: Commit**

```bash
git add backend/src/core/agentscope/events backend/tests/unit/core/agentscope/events
git commit -m "feat: add agentscope events pipeline for ag-ui redis and persistence"
```

### Task 3: Rebuild Runtime Orchestrator to Emit Internal Events

**Files:**
- Modify: `backend/src/core/agentscope/runtime/orchestrator.py`
- Modify: `backend/src/core/agentscope/runtime/__init__.py`
- Create: `backend/src/core/agentscope/runtime/agent_route_runtime.py`
- Test: `backend/tests/unit/core/agentscope/runtime/test_agent_route_runtime.py`

**Step 1: Write failing runtime tests**

```python
@pytest.mark.asyncio
async def test_runtime_emits_run_started_and_finished() -> None:
    events = await runtime.run(...)
    assert events[0].type == "run_started"
```

**Step 2: Run tests to verify failure**

Run: `uv run pytest tests/unit/core/agentscope/runtime/test_agent_route_runtime.py -q`
Expected: FAIL before runtime adapter exists.

**Step 3: Implement runtime adapter**

```python
class AgentRouteRuntime:
    async def run(self, command: RunCommand) -> RuntimeResult:
        await self._events.emit(run_started_event(...))
        ...
```

Hook existing stage runtime (intent/execution/report) and stream text/tool events into pipeline.

**Step 4: Re-run tests**

Run: `uv run pytest tests/unit/core/agentscope/runtime/test_agent_route_runtime.py -q`
Expected: PASS.

**Step 5: Commit**

```bash
git add backend/src/core/agentscope/runtime backend/tests/unit/core/agentscope/runtime/test_agent_route_runtime.py
git commit -m "feat: add agentscope runtime adapter for agent route commands"
```

### Task 4: Replace v1 Agent Service Dependencies with AgentScope

**Files:**
- Modify: `backend/src/v1/agent/dependencies.py`
- Modify: `backend/src/v1/agent/service.py`
- Modify: `backend/src/v1/agent/router.py`
- Test: `backend/tests/unit/v1/agent/test_service.py`
- Test: `backend/tests/integration/v1/agent/test_sse_flow_live.py`

**Step 1: Write failing tests for route/service integration contracts**

```python
@pytest.mark.asyncio
async def test_enqueue_run_uses_agentscope_runtime() -> None:
    resp = await service.enqueue_run(...)
    assert resp.thread_id == input.thread_id
```

**Step 2: Run tests to verify failure**

Run: `uv run pytest tests/unit/v1/agent/test_service.py -q`
Expected: FAIL before dependency rewiring.

**Step 3: Implement rewiring**

```python
service = AgentService(runtime=AgentRouteRuntime(...), events=AgentScopeEventsFacade(...))
```

Keep paths unchanged (`/runs`, `/resume`, `/events`, `/history`), keep `/transcribe` standalone.

**Step 4: Re-run tests**

Run: `uv run pytest tests/unit/v1/agent/test_service.py tests/integration/v1/agent/test_sse_flow_live.py -q`
Expected: PASS.

**Step 5: Commit**

```bash
git add backend/src/v1/agent backend/tests/unit/v1/agent backend/tests/integration/v1/agent
git commit -m "refactor: route v1 agent endpoints to agentscope runtime"
```

### Task 5: Migrate Session/Message Persistence Ownership to AgentScope Events

**Files:**
- Modify: `backend/src/models/agent_chat_session.py`
- Modify: `backend/src/models/agent_chat_message.py`
- Modify/Create migrations under `backend/alembic/versions/*`
- Create: `backend/tests/integration/core/agentscope/test_persistence_metrics.py`

**Step 1: Write failing integration tests for metrics persistence**

```python
@pytest.mark.asyncio
async def test_message_tokens_cost_latency_persisted() -> None:
    ...
    assert row.input_tokens > 0
```

**Step 2: Run tests to verify failure**

Run: `uv run pytest tests/integration/core/agentscope/test_persistence_metrics.py -q`
Expected: FAIL until event store persists metrics.

**Step 3: Implement persistence updates/migration if needed**

```python
await store.persist_message(..., input_tokens=..., latency_ms=...)
```

**Step 4: Re-run tests**

Run: `uv run pytest tests/integration/core/agentscope/test_persistence_metrics.py -q`
Expected: PASS.

**Step 5: Commit**

```bash
git add backend/src/core/agentscope/events/store.py backend/src/models backend/alembic/versions backend/tests/integration/core/agentscope/test_persistence_metrics.py
git commit -m "feat: persist agentscope session and message metrics"
```

### Task 6: Remove core/agent and Finalize Imports

**Files:**
- Delete: `backend/src/core/agent/**`
- Modify: all import sites found by grep
- Test: `backend/tests/**` impacted suites

**Step 1: Write guard tests proving no core.agent imports remain**

```python
def test_no_core_agent_imports() -> None:
    ...
```

**Step 2: Run guard test and verify failure**

Run: `uv run pytest tests/unit/core/agentscope/test_no_legacy_agent_imports.py -q`
Expected: FAIL before cleanup.

**Step 3: Remove old module and update imports**

```python
# replace from core.agent... with core.agentscope...
```

**Step 4: Run full verification**

Run:
- `uv run pytest tests/unit/core/agentscope tests/unit/v1/agent -q`
- `uv run pytest tests/integration/core/agentscope tests/integration/v1/agent -q`
- `uv run ruff check src tests`
- `uv run basedpyright src tests`

Expected: PASS.

**Step 5: Commit**

```bash
git add backend/src backend/tests
git commit -m "refactor: remove legacy core agent module after agentscope migration"
```

### Task 7: Frontend Contract Verification (No Route Change)

**Files:**
- Verify: `apps/lib/features/chat/data/models/ag_ui_event.dart`
- Verify: `apps/lib/features/chat/data/services/ag_ui_service.dart`
- Test: `apps/test/features/chat/**`

**Step 1: Add failing compatibility test for required AG-UI events**

```dart
test('supports run/text/tool event sequence') { ... }
```

**Step 2: Run test to verify failure**

Run: `cd apps && flutter test test/features/chat/...`
Expected: FAIL until backend event payload normalization is aligned.

**Step 3: Implement backend compatibility fixes only**

Keep frontend route and event type expectations unchanged where possible.

**Step 4: Re-run Flutter tests**

Run: `cd apps && flutter test`
Expected: PASS on impacted suites.

**Step 5: Commit**

```bash
git add apps/lib apps/test
git commit -m "test: verify ag-ui event contract compatibility for chat client"
```