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

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

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

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

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

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

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

@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

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

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

@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

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

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

@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

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

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

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

# 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

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

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

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