diff --git a/backend/api/server.py b/backend/api/server.py index 904d3e5..97d2ae2 100644 --- a/backend/api/server.py +++ b/backend/api/server.py @@ -71,13 +71,19 @@ from src.run_state_store import ACTIVE_RUN_STATUSES, RunStateStore from src.session_store import ( DEFAULT_AGENT_SESSION_DIR, StoredAgentSession, + agent_session_delta, + cancel_session_input_queue, + consume_session_guidance, delete_agent_session, deserialize_runtime_config, + enqueue_session_input, list_agent_sessions, + list_session_input_queue, load_agent_session, save_agent_session, serialize_model_config, serialize_runtime_config, + update_session_input_queue_item, ) from src.token_budget import calculate_token_budget @@ -1061,6 +1067,22 @@ class RunCancelRequest(BaseModel): run_id: str | None = None +class SessionInputQueueCreate(BaseModel): + content: str = Field(min_length=1, max_length=20000) + kind: str = 'next_turn' + run_id: str | None = None + account_id: str | None = None + metadata: dict[str, Any] | None = None + + +class SessionInputQueueUpdate(BaseModel): + content: str | None = Field(default=None, max_length=20000) + kind: str | None = None + status: str | None = None + run_id: str | None = None + account_id: str | None = None + + class StateUpdate(BaseModel): model: str | None = None base_url: str | None = None @@ -1737,6 +1759,126 @@ def create_app(state: AgentState) -> FastAPI: raise HTTPException(status_code=404, detail='Session not found') return _serialize_stored_session(stored) + @app.get('/api/sessions/{session_id}/state') + async def get_session_state_delta( + session_id: str, + account_id: str | None = None, + after_message_seq: int = 0, + after_event_seq: int = 0, + ) -> dict[str, Any]: + directory = state.account_paths(account_id)['sessions'] + safe_id = _safe_session_id(session_id) + if safe_id is None: + raise HTTPException(status_code=400, detail='Invalid session id') + account_key = state._account_key(account_id) + message_delta = agent_session_delta( + safe_id, + directory=directory, + after_message_seq=after_message_seq, + ) + event_delta = state.run_state_store.events_since( + account_key, + safe_id, + after_event_seq=after_event_seq, + ) + run_status = await latest_run(safe_id, account_id=account_id) + queue_items = list_session_input_queue(safe_id, directory=directory) + return { + 'session_id': safe_id, + 'session': message_delta.get('session', {'session_id': safe_id}), + 'messages': message_delta.get('messages', []), + 'latest_message_seq': message_delta.get('latest_message_seq', 0), + 'activity_events': event_delta.get('events', []), + 'latest_event_seq': event_delta.get('latest_event_seq', 0), + 'run': run_status, + 'input_queue': queue_items, + } + + @app.get('/api/sessions/{session_id}/input-queue') + async def get_session_input_queue( + session_id: str, + account_id: str | None = None, + ) -> dict[str, Any]: + directory = state.account_paths(account_id)['sessions'] + safe_id = _safe_session_id(session_id) + if safe_id is None: + raise HTTPException(status_code=400, detail='Invalid session id') + return { + 'session_id': safe_id, + 'items': list_session_input_queue(safe_id, directory=directory), + } + + @app.post('/api/sessions/{session_id}/input-queue') + async def create_session_input_queue_item( + session_id: str, + payload: SessionInputQueueCreate, + ) -> dict[str, Any]: + directory = state.account_paths(payload.account_id)['sessions'] + safe_id = _safe_session_id(session_id) + if safe_id is None: + raise HTTPException(status_code=400, detail='Invalid session id') + content = payload.content.strip() + if not content: + raise HTTPException(status_code=400, detail='Empty input content') + run_id = payload.run_id + if payload.kind == 'guidance' and not run_id: + account_key = state._account_key(payload.account_id) + latest = state.run_state_store.snapshot_latest(account_key, safe_id) + if latest and latest.get('status') in ACTIVE_RUN_STATUSES: + run_id = str(latest.get('run_id') or '') + item = enqueue_session_input( + safe_id, + directory=directory, + run_id=run_id, + kind=payload.kind, + content=content, + metadata=payload.metadata, + ) + return {'session_id': safe_id, 'item': item} + + @app.patch('/api/sessions/{session_id}/input-queue/{item_id}') + async def update_session_input_queue( + session_id: str, + item_id: int, + payload: SessionInputQueueUpdate, + ) -> dict[str, Any]: + directory = state.account_paths(payload.account_id)['sessions'] + safe_id = _safe_session_id(session_id) + if safe_id is None: + raise HTTPException(status_code=400, detail='Invalid session id') + item = update_session_input_queue_item( + safe_id, + item_id, + directory=directory, + content=payload.content.strip() if payload.content is not None else None, + kind=payload.kind, + status=payload.status, + run_id=payload.run_id, + ) + if item is None: + raise HTTPException(status_code=404, detail='Queue item not found') + return {'session_id': safe_id, 'item': item} + + @app.delete('/api/sessions/{session_id}/input-queue/{item_id}') + async def delete_session_input_queue( + session_id: str, + item_id: int, + account_id: str | None = None, + ) -> dict[str, Any]: + directory = state.account_paths(account_id)['sessions'] + safe_id = _safe_session_id(session_id) + if safe_id is None: + raise HTTPException(status_code=400, detail='Invalid session id') + item = update_session_input_queue_item( + safe_id, + item_id, + directory=directory, + status='cancelled', + ) + if item is None: + raise HTTPException(status_code=404, detail='Queue item not found') + return {'session_id': safe_id, 'item': item} + @app.delete('/api/sessions/{session_id}') async def delete_session(session_id: str, account_id: str | None = None) -> dict[str, Any]: directory = state.account_paths(account_id)['sessions'] @@ -2356,8 +2498,18 @@ def create_app(state: AgentState) -> FastAPI: status='cancelled', stage='用户已取消', ) + queue_cancelled = cancel_session_input_queue( + safe_id, + state.account_paths(payload.account_id)['sessions'], + run_id=payload.run_id, + ) cancelled_ids.update(stored_cancelled) - cancelled = bool(cancelled_ids or cancelled_bg_task_ids or abandoned_runtime) + cancelled = bool( + cancelled_ids + or cancelled_bg_task_ids + or abandoned_runtime + or queue_cancelled + ) if cancelled: _mark_session_interrupted( state.account_paths(payload.account_id)['sessions'], @@ -2370,6 +2522,7 @@ def create_app(state: AgentState) -> FastAPI: 'cancelled_run_ids': sorted(cancelled_ids), 'cancelled_bg_task_ids': sorted(cancelled_bg_task_ids), 'abandoned_runtime': abandoned_runtime, + 'cancelled_input_queue_items': queue_cancelled, } def _run_chat_payload( @@ -2527,6 +2680,22 @@ def create_app(state: AgentState) -> FastAPI: except Exception: # noqa: BLE001 return False + def _drain_runtime_guidance() -> tuple[dict[str, object], ...]: + items = consume_session_guidance( + requested_session_id, + run_record.run_id, + directory=session_directory, + ) + return tuple( + { + 'id': int(item.get('id') or 0), + 'content': str(item.get('content') or ''), + } + for item in items + ) + + previous_guidance_provider = agent.runtime_guidance_provider + agent.runtime_guidance_provider = _drain_runtime_guidance agent.tool_context = replace( agent.tool_context, cancel_event=run_record.cancel_event, @@ -2602,6 +2771,7 @@ def create_app(state: AgentState) -> FastAPI: ) raise finally: + agent.runtime_guidance_provider = previous_guidance_provider agent.tool_context = replace( agent.tool_context, cancel_event=None, diff --git a/docs/technical-architecture/12-runtime-guidance-queue.md b/docs/technical-architecture/12-runtime-guidance-queue.md new file mode 100644 index 0000000..c896e0b --- /dev/null +++ b/docs/technical-architecture/12-runtime-guidance-queue.md @@ -0,0 +1,125 @@ +# 运行中输入队列与 Runtime Guidance 注入 + +## 背景 + +用户在一个会话运行中继续输入，是 Agent 产品的基本能力。这个输入不能直接当成普通 user message 写入当前模型历史，否则会产生三个问题： + +1. **串台**：前端切换 session 或 URL 状态滞后时，新输入可能被写进旧 session。 +2. **取消误伤**：新建任务或继续输入会触发新的 run，从而取消当前正在运行的 run。 +3. **上下文污染**：运行中的输入如果直接进入 `model_messages`，会破坏当前 tool_use/tool_result 顺序，甚至触发 Bedrock/Anthropic 的 tool_result 校验错误。 + +正确做法是把运行中输入先作为 UI 和 runtime 的外部事件持久化，等 Agent loop 进入安全边界时再决定如何注入。 + +## 主流方案对比 + +| 方案 | 关键机制 | 对本项目的启发 | +|------|----------|----------------| +| [OpenAI Codex long-horizon tasks](https://developers.openai.com/blog/run-long-horizon-tasks-with-codex) | 长任务依赖计划、验证、修复和可持续的外部状态，而不是单轮大 prompt | 会话运行态要可恢复；用户中途修正不能重置整轮任务 | +| [Claude Code hooks](https://code.claude.com/docs/en/hooks-guide) | `UserPromptSubmit`、`PreToolUse`、`PostToolUse`、`Stop` 等生命周期点允许注入上下文或阻断动作 | runtime guidance 应只在明确生命周期边界注入，不直接改写当前消息流 | +| [Building AI Coding Agents for the Terminal](https://arxiv.org/html/2603.05344v1) | Agent harness 把输入层、工具层、上下文层和执行层拆开；输入可通过线程安全队列进入执行循环 | 运行中输入应先入队，再由 Agent loop 主线程消费 | +| [Event-driven agentic loops](https://boundaryml.com/podcast/2025-11-05-event-driven-agents) | 用户输入、LLM chunk、tool call、interrupt 都是事件；UI、LLM、持久化各自投影 | `display_messages`、`model_messages`、`run_events` 必须分离，避免一个状态源服务所有场景 | + +## 目标设计 + +```text +用户输入 + -> 如果当前 session idle: 正常发送，创建 run + -> 如果当前 session running: 写入 agent_input_queue + -> UI 展示 pending chip + -> 用户可编辑、删除、引导 + -> 引导: kind=guidance，绑定当前 run_id + -> Agent loop 下一轮开始前消费 guidance + -> guidance 以 display=false 的 user message 注入 model_messages +``` + +## 数据分层 + +| 数据 | 作用 | 是否允许 compact 覆盖 | +|------|------|-----------------------| +| `model_messages` | 给模型推理用，可压缩、可摘要、可隐藏注入 | 允许 | +| `display_messages` / `agent_display_messages` | 给 UI 回放用，append-only，不因为 compact 丢历史 | 不允许 | +| `run_states` | 当前 run 的状态、耗时、取消能力 | 不允许用前端内存替代 | +| `run_events` | 右侧活动区事件流 | 不允许只存在 SSE 内存里 | +| `agent_input_queue` | 运行中输入、guidance、待处理后续输入 | 不允许直接写进 display/model messages | + +## 后端实现约定 + +### 状态读取 + +前端优先读取： + +```text +GET /api/sessions/{session_id}/state +``` + +返回： + +- `messages`: DB 中 `agent_display_messages` 的增量或全量。 +- `activity_events`: DB 中 `run_events` 的增量或全量。 +- `run`: `run_states` 最新状态。 +- `input_queue`: 当前 pending 输入队列。 + +旧接口 `GET /api/sessions/{session_id}` 只作为兼容兜底，不应该再作为实时 UI 的主状态源。 + +### 输入队列 + +```text +POST /api/sessions/{session_id}/input-queue +GET /api/sessions/{session_id}/input-queue +PATCH /api/sessions/{session_id}/input-queue/{item_id} +DELETE /api/sessions/{session_id}/input-queue/{item_id} +``` + +字段约定： + +| 字段 | 说明 | +|------|------| +| `kind=next_turn` | 运行中输入的默认状态，只展示在 composer 上方，不进入模型 | +| `kind=guidance` | 用户显式点击“引导”后进入当前 run | +| `run_id` | guidance 应绑定当前 active run；未绑定时由后端尝试绑定 latest active run | +| `status=pending` | UI 可见，等待处理 | +| `status=consumed` | 已被 runtime 注入 | +| `status=cancelled` | 用户编辑/删除/取消 run 后不再处理 | + +### Runtime 注入 + +`LocalCodingAgent` 提供 `runtime_guidance_provider`，在每轮模型调用前消费 pending guidance： + +```text +agent loop boundary + -> consume_session_guidance(session_id, run_id) + -> append hidden user message: + + 用户在任务运行中补充了以下引导... + + -> display=false + -> 继续模型调用 +``` + +这个注入点必须在 tool_result 已经写回之后、下一次模型调用之前，不能插在 tool_use 和 tool_result 中间。 + +## 前端交互 + +1. 当前 session idle：输入框 Enter 仍然正常发送。 +2. 当前 session running：输入框不禁用；Enter 写入 queue，清空输入框。 +3. pending 输入显示为 composer 上方 chip： + - **编辑**：取消 queue item，把文本恢复到输入框。 + - **引导**：改成 `kind=guidance` 并绑定 active `run_id`。 + - **删除**：取消 queue item。 +4. 停止 run 时，后端同时取消该 session 下 pending queue item，避免下一轮误消费。 + +## 不做的事 + +- 不在运行中输入时自动创建新 run。 +- 不把 pending 输入直接写入 `display_messages`。 +- 不把 guidance 展示成普通用户消息；它是运行中的控制信号，不是对话历史。 +- 不依赖前端内存判断最终状态；刷新后必须能从 DB 完整恢复。 + +## 验收点 + +1. 同一个账号同时打开两个 session，分别运行任务，输入不会串台。 +2. A session 运行中切到 B session 输入，B 的输入只进入 B 的 queue。 +3. A session 运行中输入后刷新，pending chip 仍存在。 +4. 点击“引导”后，下一次模型调用前能消费 guidance，并在活动区记录注入事件。 +5. 点击停止后，对应 run 的进程和 pending queue 都被取消。 +6. 触发 compact 后，UI 仍能从 `display_messages` 回放完整历史；模型只使用 compact 后的 `model_messages`。 diff --git a/docs/technical-architecture/README.md b/docs/technical-architecture/README.md index 8e713bd..979ba12 100644 --- a/docs/technical-architecture/README.md +++ b/docs/technical-architecture/README.md @@ -15,6 +15,7 @@ 8. [label-master Skill 实现](08-label-master.md) 9. [外部系统 Skill：ELK、SQL、模型迭代](09-external-skills.md) 10. [Agent 记忆机制调研与对比](10-memory-research.md) +11. [运行中输入队列与 Runtime Guidance 注入](12-runtime-guidance-queue.md) ## 一句话定位 diff --git a/frontend/app/app/api/claw/sessions/[sessionId]/input-queue/[itemId]/route.ts b/frontend/app/app/api/claw/sessions/[sessionId]/input-queue/[itemId]/route.ts new file mode 100644 index 0000000..6c36f03 --- /dev/null +++ b/frontend/app/app/api/claw/sessions/[sessionId]/input-queue/[itemId]/route.ts @@ -0,0 +1,60 @@ +import { getCurrentAccount } from "@/lib/claw-auth"; + +const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765"; + +export async function PATCH( + req: Request, + { params }: { params: Promise<{ sessionId: string; itemId: string }> }, +) { + const account = await getCurrentAccount(); + if (!account) + return Response.json({ error: "请先登录账号" }, { status: 401 }); + + const { sessionId: rawSessionId, itemId: rawItemId } = await params; + const sessionId = rawSessionId.trim(); + const itemId = rawItemId.trim(); + const payload = await req.json().catch(() => ({})); + const url = new URL( + `${CLAW_API_URL}/api/sessions/${sessionId}/input-queue/${itemId}`, + ); + const response = await fetch(url, { + method: "PATCH", + cache: "no-store", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ ...payload, account_id: account.id }), + }); + const responsePayload = await response.text(); + return new Response(responsePayload, { + status: response.status, + headers: { + "content-type": + response.headers.get("content-type") ?? "application/json", + }, + }); +} + +export async function DELETE( + _req: Request, + { params }: { params: Promise<{ sessionId: string; itemId: string }> }, +) { + const account = await getCurrentAccount(); + if (!account) + return Response.json({ error: "请先登录账号" }, { status: 401 }); + + const { sessionId: rawSessionId, itemId: rawItemId } = await params; + const sessionId = rawSessionId.trim(); + const itemId = rawItemId.trim(); + const url = new URL( + `${CLAW_API_URL}/api/sessions/${sessionId}/input-queue/${itemId}`, + ); + url.searchParams.set("account_id", account.id); + const response = await fetch(url, { method: "DELETE", cache: "no-store" }); + const responsePayload = await response.text(); + return new Response(responsePayload, { + status: response.status, + headers: { + "content-type": + response.headers.get("content-type") ?? "application/json", + }, + }); +} diff --git a/frontend/app/app/api/claw/sessions/[sessionId]/input-queue/route.ts b/frontend/app/app/api/claw/sessions/[sessionId]/input-queue/route.ts new file mode 100644 index 0000000..2ae301f --- /dev/null +++ b/frontend/app/app/api/claw/sessions/[sessionId]/input-queue/route.ts @@ -0,0 +1,54 @@ +import { getCurrentAccount } from "@/lib/claw-auth"; + +const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765"; + +export async function GET( + _req: Request, + { params }: { params: Promise<{ sessionId: string }> }, +) { + const account = await getCurrentAccount(); + if (!account) + return Response.json({ error: "请先登录账号" }, { status: 401 }); + + const { sessionId: rawSessionId } = await params; + const sessionId = rawSessionId.trim(); + const url = new URL(`${CLAW_API_URL}/api/sessions/${sessionId}/input-queue`); + url.searchParams.set("account_id", account.id); + const response = await fetch(url, { cache: "no-store" }); + const payload = await response.text(); + return new Response(payload, { + status: response.status, + headers: { + "content-type": + response.headers.get("content-type") ?? "application/json", + }, + }); +} + +export async function POST( + req: Request, + { params }: { params: Promise<{ sessionId: string }> }, +) { + const account = await getCurrentAccount(); + if (!account) + return Response.json({ error: "请先登录账号" }, { status: 401 }); + + const { sessionId: rawSessionId } = await params; + const sessionId = rawSessionId.trim(); + const payload = await req.json().catch(() => ({})); + const url = new URL(`${CLAW_API_URL}/api/sessions/${sessionId}/input-queue`); + const response = await fetch(url, { + method: "POST", + cache: "no-store", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ ...payload, account_id: account.id }), + }); + const responsePayload = await response.text(); + return new Response(responsePayload, { + status: response.status, + headers: { + "content-type": + response.headers.get("content-type") ?? "application/json", + }, + }); +} diff --git a/frontend/app/app/api/claw/sessions/[sessionId]/state/route.ts b/frontend/app/app/api/claw/sessions/[sessionId]/state/route.ts new file mode 100644 index 0000000..fac6bff --- /dev/null +++ b/frontend/app/app/api/claw/sessions/[sessionId]/state/route.ts @@ -0,0 +1,31 @@ +import { getCurrentAccount } from "@/lib/claw-auth"; + +const CLAW_API_URL = process.env.CLAW_API_URL ?? "http://127.0.0.1:8765"; + +export async function GET( + req: Request, + { params }: { params: Promise<{ sessionId: string }> }, +) { + const account = await getCurrentAccount(); + if (!account) + return Response.json({ error: "请先登录账号" }, { status: 401 }); + + const { sessionId: rawSessionId } = await params; + const sessionId = rawSessionId.trim(); + const incoming = new URL(req.url); + const url = new URL(`${CLAW_API_URL}/api/sessions/${sessionId}/state`); + url.searchParams.set("account_id", account.id); + for (const key of ["after_message_seq", "after_event_seq"]) { + const value = incoming.searchParams.get(key); + if (value) url.searchParams.set(key, value); + } + const response = await fetch(url, { cache: "no-store" }); + const payload = await response.text(); + return new Response(payload, { + status: response.status, + headers: { + "content-type": + response.headers.get("content-type") ?? "application/json", + }, + }); +} diff --git a/frontend/app/app/doc/[slug]/page.tsx b/frontend/app/app/doc/[slug]/page.tsx index c3f7b2c..5caeeba 100644 --- a/frontend/app/app/doc/[slug]/page.tsx +++ b/frontend/app/app/doc/[slug]/page.tsx @@ -61,6 +61,11 @@ const technicalDocs = { file: "11-subagents.md", image: null, }, + "12-runtime-guidance-queue": { + title: "运行中输入队列", + file: "12-runtime-guidance-queue.md", + image: null, + }, } as const; type TechnicalDocSlug = keyof typeof technicalDocs; diff --git a/frontend/app/app/doc/page.tsx b/frontend/app/app/doc/page.tsx index 0a1161f..7f3d1a6 100644 --- a/frontend/app/app/doc/page.tsx +++ b/frontend/app/app/doc/page.tsx @@ -144,6 +144,27 @@ const sections: Section[] = [ decision: "子 Agent 负责观察和验证，主 Agent 负责用户上下文、最终决策和产物写入。", }, + { + id: "runtime-guidance", + no: "12", + title: "运行中输入队列", + summary: + "运行中的用户输入先入队，只有用户显式引导时才在 Agent loop 安全边界注入当前 run。", + doc: "/doc/12-runtime-guidance-queue", + points: [ + "display_messages、model_messages、run_events 和 input_queue 分离，避免刷新、compact 和多会话切换互相污染。", + "运行中 Enter 写入队列，不创建新 run，不取消当前 run。", + "引导消息以 hidden runtime guidance 注入模型，不作为普通用户历史展示。", + ], + code: [ + "src/session_store.py", + "src/run_state_store.py", + "src/agent_runtime.py", + "frontend input queue chips", + ], + decision: + "UI、模型和持久化不能共享同一份 mutable messages；运行中交互必须事件化、可恢复、可取消。", + }, ]; const skillSections: Section[] = [ @@ -235,7 +256,8 @@ const skillSections: Section[] = [ const exampleSessions = [ { title: "product-data", - description: "数据生成链路：目标抽取、计划确认、分批生成、canonical records 和训练/评测格式导出。", + description: + "数据生成链路：目标抽取、计划确认、分批生成、canonical records 和训练/评测格式导出。", links: [ "http://10.189.47.6/session/__LOCALID_ZgwXdfP", "http://10.189.47.6/session/__LOCALID_QOecDTu", @@ -243,7 +265,8 @@ const exampleSessions = [ }, { title: "标签大师", - description: "标签知识判断链路：复杂度、多指令、自动任务、function/tag 输出形态和边界知识引用。", + description: + "标签知识判断链路：复杂度、多指令、自动任务、function/tag 输出形态和边界知识引用。", links: [ "http://10.189.47.6/session/__LOCALID_ji5lCfb", "http://10.189.47.6/session/__LOCALID_Z22r3pn", @@ -254,12 +277,14 @@ const exampleSessions = [ }, { title: "标签大师 + online-mining", - description: "组合链路：先用线上日志定位样本，再结合标签知识做判断、整理和格式转换。", + description: + "组合链路：先用线上日志定位样本，再结合标签知识做判断、整理和格式转换。", links: ["http://10.189.47.6/session/__LOCALID_OLKxKnf"], }, { title: "online-mining", - description: "线上挖掘链路：查询策略、ELK 字段探索、候选样本抽样 review 和标准数据沉淀。", + description: + "线上挖掘链路：查询策略、ELK 字段探索、候选样本抽样 review 和标准数据沉淀。", links: [ "http://10.189.47.6/session/__LOCALID_BnxMLqn", "http://10.189.47.6/session/__LOCALID_xZRilR5", @@ -357,7 +382,8 @@ export default function DocPage() {

ZK-Data-Agent 架构

- 基座是一个通用 Code Agent Loop；我们的工作不是重写这个闭环，而是在关键节点补上团队业务能力，让执行、业务上下文和流程沉淀进入同一套工作台。 + 基座是一个通用 Code Agent + Loop；我们的工作不是重写这个闭环，而是在关键节点补上团队业务能力，让执行、业务上下文和流程沉淀进入同一套工作台。

@@ -379,19 +405,23 @@ export default function DocPage() {

基座提供执行闭环

- Code Agent 负责把任务放进多轮循环：组装上下文、模型推理、调用工具、观察结果，再继续规划或完成。它让模型从一次回答变成可以持续执行的运行时。 + Code Agent + 负责把任务放进多轮循环：组装上下文、模型推理、调用工具、观察结果，再继续规划或完成。它让模型从一次回答变成可以持续执行的运行时。

增强层体现我们的工作

- ZK-Data-Agent 在上下文、工具、产物和团队协作节点接入 Skill 体系、Skill 热更新、用户记忆、Skill 记忆、portable scripts、会话工作区、活动流和管理后台。 + ZK-Data-Agent 在上下文、工具、产物和团队协作节点接入 Skill + 体系、Skill 热更新、用户记忆、Skill 记忆、portable + scripts、会话工作区、活动流和管理后台。

效率来自工程化补齐

- 工具和工作区提升研发执行效率；Skill 与记忆提升业务嵌入深度；Git 化 Skill、热更新和团队治理提升流程可复用度。 + 工具和工作区提升研发执行效率；Skill 与记忆提升业务嵌入深度；Git 化 + Skill、热更新和团队治理提升流程可复用度。

@@ -411,7 +441,9 @@ export default function DocPage() {

示例 Session

- 下面这些是真实跑过的会话，适合讲解时直接打开。它们展示的不是单次问答，而是 Skill 如何在 Agent Loop 中完成计划、工具调用、人工确认、产物生成和格式转换。 + 下面这些是真实跑过的会话，适合讲解时直接打开。它们展示的不是单次问答，而是 + Skill 如何在 Agent Loop + 中完成计划、工具调用、人工确认、产物生成和格式转换。

diff --git a/frontend/app/components/assistant-ui/thread.tsx b/frontend/app/components/assistant-ui/thread.tsx index fa64a7e..2798255 100644 --- a/frontend/app/components/assistant-ui/thread.tsx +++ b/frontend/app/components/assistant-ui/thread.tsx @@ -91,12 +91,14 @@ import { writeActiveSessionId, writePendingWorkspaceSessionId, } from "@/lib/claw-active-session"; +import { fetchSessionReplaySnapshot } from "@/lib/claw-session-cache"; import { useClawSessionReplay } from "@/lib/claw-session-replay"; import { readSessionIdFromUrl } from "@/lib/claw-session-url"; import { dispatchSkillToggled } from "@/lib/use-training-mode"; import { cn } from "@/lib/utils"; type ComposerInsertEvent = CustomEvent<{ text: string }>; +const INPUT_QUEUE_CHANGED_EVENT = "claw-input-queue-changed"; type SkillSummary = { name: string; @@ -129,11 +131,23 @@ type ClawState = { active_session_id?: string | null; }; -type ClawStoredSession = { - session_id?: string; +type ClawStoredSession = Parameters[0] & { model?: string; }; +type SessionInputQueueItem = { + id: number; + session_id: string; + run_id?: string; + kind: "next_turn" | "guidance"; + status: "pending" | "consumed" | "cancelled"; + content: string; + created_at?: number; + updated_at?: number; + consumed_at?: number | null; + metadata?: Record; +}; + type ModelOption = { id: string; provider?: string; @@ -220,32 +234,27 @@ function useRefreshReplayedRun() { if (!replayedRun.active || !replayedRun.sessionId) return; const activeSessionId = replayedRun.sessionId; let cancelled = false; + let lastSignature: string | null = null; async function refreshIfFinished() { - const runStatus = await fetchLatestRunStatus(activeSessionId); - if (cancelled) return; - if (runStatus?.status === "queued" || runStatus?.status === "running") { - const response = await fetch(`/api/claw/sessions/${activeSessionId}`, { - cache: "no-store", - }); - if (!response.ok || cancelled) return; - const payload = await response.json(); - replaySession( - activeSessionId, - toReplayRepository(payload, activeSessionId, runStatus), - ); - return; - } - const response = await fetch(`/api/claw/sessions/${activeSessionId}`, { - cache: "no-store", - }); - if (!response.ok || cancelled) return; - const payload = await response.json(); + const snapshot = await fetchSessionReplaySnapshot< + ClawStoredSession, + ClawRunStatus + >(activeSessionId); + if (cancelled || !snapshot) return; + if (snapshot.signature === lastSignature) return; + lastSignature = snapshot.signature; replaySession( activeSessionId, - toReplayRepository(payload, activeSessionId, runStatus), + toReplayRepository( + snapshot.session, + activeSessionId, + snapshot.runStatus, + ), ); - window.dispatchEvent(new Event("claw-sessions-changed")); + if (!isActiveRunStatus(snapshot.runStatus)) { + window.dispatchEvent(new Event("claw-sessions-changed")); + } } refreshIfFinished(); @@ -279,7 +288,11 @@ function useRefreshCurrentRun() { async function refreshIfSettled() { if (!sessionId) return; - const runStatus = await fetchLatestRunStatus(sessionId); + const snapshot = await fetchSessionReplaySnapshot< + ClawStoredSession, + ClawRunStatus + >(sessionId); + const runStatus = snapshot?.runStatus ?? null; if (cancelled) return; if (runStatus?.status === "queued" || runStatus?.status === "running") { const runKey = runStatus.run_id ?? "active"; @@ -291,14 +304,10 @@ function useRefreshCurrentRun() { // useRefreshReplayedRun will take over polling once replayedRun.active // becomes true. if (!runtimeRunning && previousRunKey !== runKey) { - const response = await fetch(`/api/claw/sessions/${sessionId}`, { - cache: "no-store", - }); - if (!response.ok || cancelled) return; - const payload = await response.json(); + if (!snapshot || cancelled) return; replaySession( sessionId, - toReplayRepository(payload, sessionId, runStatus), + toReplayRepository(snapshot.session, sessionId, runStatus), ); } return; @@ -312,14 +321,10 @@ function useRefreshCurrentRun() { runStatus?.elapsed_ms ?? "", ].join(":"); if (appliedTerminalRefMap.current.get(sessionId) === terminalKey) return; - const response = await fetch(`/api/claw/sessions/${sessionId}`, { - cache: "no-store", - }); - if (!response.ok || cancelled) return; - const payload = await response.json(); + if (!snapshot || cancelled) return; replaySession( sessionId, - toReplayRepository(payload, sessionId, runStatus), + toReplayRepository(snapshot.session, sessionId, runStatus), ); activeRunRefMap.current.delete(sessionId); appliedTerminalRefMap.current.set(sessionId, terminalKey); @@ -893,15 +898,10 @@ const ThreadSuggestionItem: FC = () => { }; const Composer: FC = () => { - const replayedRun = useReplayedRunState(); const workspaceSessionId = useCurrentThreadSessionId({ includePending: true, includeActive: true, }); - const replayedRunIsCurrent = - replayedRun.active && - Boolean(workspaceSessionId) && - replayedRun.sessionId === workspaceSessionId; return ( @@ -911,13 +911,13 @@ const Composer: FC = () => { > + @@ -926,6 +926,92 @@ const Composer: FC = () => { ); }; +const ComposerPendingInputQueue: FC<{ sessionId: string | null }> = ({ + sessionId, +}) => { + const backendRunStatus = useBackendActiveRunStatus(sessionId); + const items = useSessionInputQueue(sessionId); + if (!sessionId || items.length === 0) return null; + + return ( +
+ {items.map((item) => { + const isGuidance = item.kind === "guidance"; + const canGuide = Boolean(backendRunStatus?.run_id); + return ( +
+ + + {isGuidance ? "待引导" : "待发送"} + + {item.content} + + + + +
+ ); + })} +
+ ); +}; + const ComposerWorkspaceStatus: FC<{ sessionId: string | null }> = ({ sessionId, }) => { @@ -1071,6 +1157,100 @@ function isActiveRunStatus(status?: ClawRunStatus | null) { return status?.status === "queued" || status?.status === "running"; } +function dispatchInputQueueChanged() { + window.dispatchEvent(new Event(INPUT_QUEUE_CHANGED_EVENT)); +} + +function useSessionInputQueue(sessionId: string | null) { + const [items, setItems] = useState([]); + + useEffect(() => { + if (!sessionId) { + setItems([]); + return; + } + let cancelled = false; + const refresh = async () => { + const nextItems = await fetchSessionInputQueue(sessionId); + if (!cancelled) setItems(nextItems); + }; + const handleQueueChanged = () => { + void refresh(); + }; + void refresh(); + const interval = window.setInterval(refresh, 1000); + window.addEventListener(INPUT_QUEUE_CHANGED_EVENT, handleQueueChanged); + return () => { + cancelled = true; + window.clearInterval(interval); + window.removeEventListener(INPUT_QUEUE_CHANGED_EVENT, handleQueueChanged); + }; + }, [sessionId]); + + return items; +} + +async function fetchSessionInputQueue(sessionId: string) { + try { + const response = await fetch( + `/api/claw/sessions/${encodeURIComponent(sessionId)}/input-queue`, + { cache: "no-store" }, + ); + if (!response.ok) return []; + const payload = (await response.json()) as { + items?: SessionInputQueueItem[]; + }; + return Array.isArray(payload.items) ? payload.items : []; + } catch { + return []; + } +} + +async function createSessionInputQueueItem( + sessionId: string, + payload: { + content: string; + kind?: "next_turn" | "guidance"; + run_id?: string; + }, +) { + const response = await fetch( + `/api/claw/sessions/${encodeURIComponent(sessionId)}/input-queue`, + { + method: "POST", + cache: "no-store", + headers: { "content-type": "application/json" }, + body: JSON.stringify(payload), + }, + ); + if (!response.ok) { + throw new Error(`Failed to queue input: ${response.status}`); + } + return (await response.json()) as { item?: SessionInputQueueItem }; +} + +async function updateSessionInputQueueItem( + sessionId: string, + itemId: number, + payload: Partial< + Pick + >, +) { + const response = await fetch( + `/api/claw/sessions/${encodeURIComponent(sessionId)}/input-queue/${itemId}`, + { + method: "PATCH", + cache: "no-store", + headers: { "content-type": "application/json" }, + body: JSON.stringify(payload), + }, + ); + if (!response.ok) { + throw new Error(`Failed to update queued input: ${response.status}`); + } + return (await response.json()) as { item?: SessionInputQueueItem }; +} + const ReplayedRunCancelButton: FC<{ sessionId: string | null; runId: string | null; @@ -1090,15 +1270,18 @@ const ReplayedRunCancelButton: FC<{ setCancelling(true); try { await cancelLatestRun(sessionId, runId); - const runStatus = await fetchLatestRunStatus(sessionId); - const response = await fetch(`/api/claw/sessions/${sessionId}`, { - cache: "no-store", - }); - if (response.ok) { - const payload = await response.json(); + const snapshot = await fetchSessionReplaySnapshot< + ClawStoredSession, + ClawRunStatus + >(sessionId); + if (snapshot) { replaySession( sessionId, - toReplayRepository(payload, sessionId, runStatus), + toReplayRepository( + snapshot.session, + sessionId, + snapshot.runStatus, + ), ); } window.dispatchEvent(new Event("claw-sessions-changed")); @@ -2099,10 +2282,7 @@ const ActivityChainSummary: FC<{ liveStartedAtRef.current = authoritativeStartedAt; } else if (liveStartedAtRef.current === null && storedElapsedMs !== null) { liveStartedAtRef.current = Date.now() - storedElapsedMs; - } else if ( - liveStartedAtRef.current !== null && - storedElapsedMs !== null - ) { + } else if (liveStartedAtRef.current !== null && storedElapsedMs !== null) { liveStartedAtRef.current = Math.min( liveStartedAtRef.current, Date.now() - storedElapsedMs, @@ -2422,14 +2602,25 @@ const ImeComposerInput: FC> = ({ const storeText = useAuiState((s) => s.composer.isEditing ? s.composer.text : "", ); + const isEditingComposer = useAuiState((s) => s.composer.isEditing); + const runtimeRunning = useAuiState((s) => s.thread.isRunning); const runtimeDisabled = useAuiState( (s) => s.thread.isDisabled || s.composer.dictation?.inputDisabled, ); + const activeSessionId = useCurrentThreadSessionId({ + includePending: true, + includeActive: true, + }); + const backendRunStatus = useBackendActiveRunStatus(activeSessionId); const [localText, setLocalText] = useState(storeText); const textareaRef = useRef(null); const isComposingRef = useRef(false); const justSubmittedRef = useRef(false); - const isDisabled = runtimeDisabled || disabled; + const canQueueWhileRunning = + !isEditingComposer && + Boolean(activeSessionId) && + (runtimeRunning || isActiveRunStatus(backendRunStatus)); + const isDisabled = (runtimeDisabled && !canQueueWhileRunning) || disabled; // Pull store→local on external store changes (session switch, paste-from-attachment, // submit-clear, etc). The reverse direction (local→store) is pushed eagerly by @@ -2475,6 +2666,12 @@ const ImeComposerInput: FC> = ({ const text = detail?.text ?? ""; justSubmittedRef.current = false; setLocalText(text); + requestAnimationFrame(() => { + const textarea = textareaRef.current; + if (!textarea) return; + textarea.focus({ preventScroll: true }); + textarea.setSelectionRange(text.length, text.length); + }); }; window.addEventListener("claw-composer-restore-draft", handler); return () => @@ -2523,6 +2720,27 @@ const ImeComposerInput: FC> = ({ } const threadState = aui.thread().getState(); + const content = localText.trim(); + if ( + (threadState.isRunning || isActiveRunStatus(backendRunStatus)) && + activeSessionId + ) { + event.preventDefault(); + if (!content) return; + justSubmittedRef.current = true; + setLocalText(""); + syncComposerText(""); + void createSessionInputQueueItem(activeSessionId, { + content, + kind: "next_turn", + }) + .catch(() => { + justSubmittedRef.current = false; + setLocalText(content); + }) + .finally(dispatchInputQueueChanged); + return; + } if (threadState.isRunning && !threadState.capabilities.queue) return; event.preventDefault(); diff --git a/frontend/app/lib/claw-session-cache.ts b/frontend/app/lib/claw-session-cache.ts index e431d79..f63754f 100644 --- a/frontend/app/lib/claw-session-cache.ts +++ b/frontend/app/lib/claw-session-cache.ts @@ -33,6 +33,12 @@ export async function fetchSessionReplaySnapshot< sessionId: string, options: { signal?: AbortSignal } = {}, ): Promise | null> { + const stateSnapshot = await fetchSessionStateSnapshot( + sessionId, + options.signal, + ); + if (stateSnapshot) return stateSnapshot; + const [sessionResponse, runStatus] = await Promise.all([ fetch(`/api/claw/sessions/${encodeURIComponent(sessionId)}`, { cache: "no-store", @@ -45,6 +51,43 @@ export async function fetchSessionReplaySnapshot< return writeCachedSessionReplay(sessionId, session, runStatus); } +type SessionStateResponse = { + session?: Record; + messages?: Array<{ seq?: number; message?: unknown }>; + run?: TRun | null; +}; + +async function fetchSessionStateSnapshot( + sessionId: string, + signal?: AbortSignal, +): Promise | null> { + try { + const response = await fetch( + `/api/claw/sessions/${encodeURIComponent(sessionId)}/state?after_message_seq=0&after_event_seq=0`, + { cache: "no-store", signal }, + ); + if (!response.ok) return null; + const payload = (await response.json()) as SessionStateResponse; + if (!Array.isArray(payload.messages)) return null; + const orderedMessages = payload.messages + .slice() + .sort((left, right) => (left.seq ?? 0) - (right.seq ?? 0)) + .map((entry) => entry.message) + .filter((message) => Boolean(message)); + const session = { + ...(payload.session ?? {}), + session_id: sessionId, + messages: orderedMessages, + } as TSession; + return writeCachedSessionReplay(sessionId, session, payload.run ?? null); + } catch (error) { + if (error instanceof DOMException && error.name === "AbortError") { + throw error; + } + return null; + } +} + function writeCachedSessionReplay( sessionId: string, session: TSession, @@ -84,15 +127,18 @@ async function fetchLatestRunStatusForCache( } } -function sessionReplaySignature(session: unknown, runStatus: unknown) { +function sessionReplaySignature(session: unknown, runStatusValue: unknown) { const sessionObject = isObject(session) ? session : {}; - const runObject = isObject(runStatus) ? runStatus : {}; + const runObject = isObject(runStatusValue) ? runStatusValue : {}; const messages = Array.isArray(sessionObject.messages) ? sessionObject.messages : []; const lastMessage = messages.at(-1); const lastObject = isObject(lastMessage) ? lastMessage : {}; const metadata = isObject(lastObject.metadata) ? lastObject.metadata : {}; + const runStatus = + typeof runObject.status === "string" ? runObject.status : ""; + const runIsActive = runStatus === "queued" || runStatus === "running"; return [ sessionObject.session_id, sessionObject.turns, @@ -103,10 +149,11 @@ function sessionReplaySignature(session: unknown, runStatus: unknown) { typeof lastObject.content === "string" ? lastObject.content.length : "", metadata.elapsed_ms, runObject.run_id, - runObject.status, - runObject.updated_at, + runStatus, + runObject.current_stage, + runIsActive ? "" : runObject.updated_at, runObject.finished_at, - runObject.elapsed_ms, + runIsActive ? "" : runObject.elapsed_ms, ].join("|"); } diff --git a/frontend/app/public/doc-assets/technical-docs/12-runtime-guidance-queue.md b/frontend/app/public/doc-assets/technical-docs/12-runtime-guidance-queue.md new file mode 100644 index 0000000..c896e0b --- /dev/null +++ b/frontend/app/public/doc-assets/technical-docs/12-runtime-guidance-queue.md @@ -0,0 +1,125 @@ +# 运行中输入队列与 Runtime Guidance 注入 + +## 背景 + +用户在一个会话运行中继续输入，是 Agent 产品的基本能力。这个输入不能直接当成普通 user message 写入当前模型历史，否则会产生三个问题： + +1. **串台**：前端切换 session 或 URL 状态滞后时，新输入可能被写进旧 session。 +2. **取消误伤**：新建任务或继续输入会触发新的 run，从而取消当前正在运行的 run。 +3. **上下文污染**：运行中的输入如果直接进入 `model_messages`，会破坏当前 tool_use/tool_result 顺序，甚至触发 Bedrock/Anthropic 的 tool_result 校验错误。 + +正确做法是把运行中输入先作为 UI 和 runtime 的外部事件持久化，等 Agent loop 进入安全边界时再决定如何注入。 + +## 主流方案对比 + +| 方案 | 关键机制 | 对本项目的启发 | +|------|----------|----------------| +| [OpenAI Codex long-horizon tasks](https://developers.openai.com/blog/run-long-horizon-tasks-with-codex) | 长任务依赖计划、验证、修复和可持续的外部状态，而不是单轮大 prompt | 会话运行态要可恢复；用户中途修正不能重置整轮任务 | +| [Claude Code hooks](https://code.claude.com/docs/en/hooks-guide) | `UserPromptSubmit`、`PreToolUse`、`PostToolUse`、`Stop` 等生命周期点允许注入上下文或阻断动作 | runtime guidance 应只在明确生命周期边界注入，不直接改写当前消息流 | +| [Building AI Coding Agents for the Terminal](https://arxiv.org/html/2603.05344v1) | Agent harness 把输入层、工具层、上下文层和执行层拆开；输入可通过线程安全队列进入执行循环 | 运行中输入应先入队，再由 Agent loop 主线程消费 | +| [Event-driven agentic loops](https://boundaryml.com/podcast/2025-11-05-event-driven-agents) | 用户输入、LLM chunk、tool call、interrupt 都是事件；UI、LLM、持久化各自投影 | `display_messages`、`model_messages`、`run_events` 必须分离，避免一个状态源服务所有场景 | + +## 目标设计 + +```text +用户输入 + -> 如果当前 session idle: 正常发送，创建 run + -> 如果当前 session running: 写入 agent_input_queue + -> UI 展示 pending chip + -> 用户可编辑、删除、引导 + -> 引导: kind=guidance，绑定当前 run_id + -> Agent loop 下一轮开始前消费 guidance + -> guidance 以 display=false 的 user message 注入 model_messages +``` + +## 数据分层 + +| 数据 | 作用 | 是否允许 compact 覆盖 | +|------|------|-----------------------| +| `model_messages` | 给模型推理用，可压缩、可摘要、可隐藏注入 | 允许 | +| `display_messages` / `agent_display_messages` | 给 UI 回放用，append-only，不因为 compact 丢历史 | 不允许 | +| `run_states` | 当前 run 的状态、耗时、取消能力 | 不允许用前端内存替代 | +| `run_events` | 右侧活动区事件流 | 不允许只存在 SSE 内存里 | +| `agent_input_queue` | 运行中输入、guidance、待处理后续输入 | 不允许直接写进 display/model messages | + +## 后端实现约定 + +### 状态读取 + +前端优先读取： + +```text +GET /api/sessions/{session_id}/state +``` + +返回： + +- `messages`: DB 中 `agent_display_messages` 的增量或全量。 +- `activity_events`: DB 中 `run_events` 的增量或全量。 +- `run`: `run_states` 最新状态。 +- `input_queue`: 当前 pending 输入队列。 + +旧接口 `GET /api/sessions/{session_id}` 只作为兼容兜底，不应该再作为实时 UI 的主状态源。 + +### 输入队列 + +```text +POST /api/sessions/{session_id}/input-queue +GET /api/sessions/{session_id}/input-queue +PATCH /api/sessions/{session_id}/input-queue/{item_id} +DELETE /api/sessions/{session_id}/input-queue/{item_id} +``` + +字段约定： + +| 字段 | 说明 | +|------|------| +| `kind=next_turn` | 运行中输入的默认状态，只展示在 composer 上方，不进入模型 | +| `kind=guidance` | 用户显式点击“引导”后进入当前 run | +| `run_id` | guidance 应绑定当前 active run；未绑定时由后端尝试绑定 latest active run | +| `status=pending` | UI 可见，等待处理 | +| `status=consumed` | 已被 runtime 注入 | +| `status=cancelled` | 用户编辑/删除/取消 run 后不再处理 | + +### Runtime 注入 + +`LocalCodingAgent` 提供 `runtime_guidance_provider`，在每轮模型调用前消费 pending guidance： + +```text +agent loop boundary + -> consume_session_guidance(session_id, run_id) + -> append hidden user message: + + 用户在任务运行中补充了以下引导... + + -> display=false + -> 继续模型调用 +``` + +这个注入点必须在 tool_result 已经写回之后、下一次模型调用之前，不能插在 tool_use 和 tool_result 中间。 + +## 前端交互 + +1. 当前 session idle：输入框 Enter 仍然正常发送。 +2. 当前 session running：输入框不禁用；Enter 写入 queue，清空输入框。 +3. pending 输入显示为 composer 上方 chip： + - **编辑**：取消 queue item，把文本恢复到输入框。 + - **引导**：改成 `kind=guidance` 并绑定 active `run_id`。 + - **删除**：取消 queue item。 +4. 停止 run 时，后端同时取消该 session 下 pending queue item，避免下一轮误消费。 + +## 不做的事 + +- 不在运行中输入时自动创建新 run。 +- 不把 pending 输入直接写入 `display_messages`。 +- 不把 guidance 展示成普通用户消息；它是运行中的控制信号，不是对话历史。 +- 不依赖前端内存判断最终状态；刷新后必须能从 DB 完整恢复。 + +## 验收点 + +1. 同一个账号同时打开两个 session，分别运行任务，输入不会串台。 +2. A session 运行中切到 B session 输入，B 的输入只进入 B 的 queue。 +3. A session 运行中输入后刷新，pending chip 仍存在。 +4. 点击“引导”后，下一次模型调用前能消费 guidance，并在活动区记录注入事件。 +5. 点击停止后，对应 run 的进程和 pending queue 都被取消。 +6. 触发 compact 后，UI 仍能从 `display_messages` 回放完整历史；模型只使用 compact 后的 `model_messages`。 diff --git a/src/agent_runtime.py b/src/agent_runtime.py index 2e79467..81a5bfe 100644 --- a/src/agent_runtime.py +++ b/src/agent_runtime.py @@ -298,6 +298,7 @@ class LocalCodingAgent: team_runtime: TeamRuntime | None = None workflow_runtime: WorkflowRuntime | None = None worktree_runtime: WorktreeRuntime | None = None + runtime_guidance_provider: Callable[[], tuple[dict[str, object], ...]] | None = None last_session: AgentSessionState | None = field(default=None, init=False, repr=False) last_run_result: AgentRunResult | None = field(default=None, init=False, repr=False) cumulative_usage: UsageStats = field(default_factory=UsageStats, init=False, repr=False) @@ -734,6 +735,11 @@ class LocalCodingAgent: return result for turn_index in range(1, self.runtime_config.max_turns + 1): + self._inject_runtime_guidance( + session, + stream_events, + turn_index=turn_index, + ) self._microcompact_session_if_needed( session, stream_events, @@ -4274,6 +4280,69 @@ class LocalCodingAgent: transcript=session.transcript(), ) + def _inject_runtime_guidance( + self, + session: AgentSessionState, + stream_events: list[dict[str, object]], + *, + turn_index: int, + ) -> None: + provider = self.runtime_guidance_provider + if provider is None: + return + try: + items = tuple( + item + for item in provider() + if isinstance(item, dict) + and str(item.get('content') or '').strip() + ) + except Exception as exc: # noqa: BLE001 + stream_events.append( + { + 'type': 'runtime_guidance_error', + 'turn_index': turn_index, + 'error': str(exc)[:500], + } + ) + return + if not items: + return + lines = [ + '', + ( + '用户在本轮任务运行过程中追加了以下引导。请在不丢弃当前进展的前提下' + '吸收这些要求；如果它们与当前目标冲突，先说明冲突并选择最合理的继续方式。' + ), + ] + item_ids: list[int] = [] + for index, item in enumerate(items, start=1): + item_id = item.get('id') + if isinstance(item_id, int) and not isinstance(item_id, bool): + item_ids.append(item_id) + content = str(item.get('content') or '').strip() + lines.append(f'{index}. {content}') + lines.append('') + session.append_user( + '\n'.join(lines), + metadata={ + 'kind': 'runtime_guidance', + 'turn_index': turn_index, + 'queue_item_ids': item_ids, + }, + message_id=f'runtime_guidance_{turn_index}_{uuid4().hex[:8]}', + display=False, + ) + stream_events.append( + { + 'type': 'runtime_guidance_injected', + 'turn_index': turn_index, + 'input_ids': item_ids, + 'count': len(items), + 'message': f'已将 {len(items)} 条用户引导注入当前任务', + } + ) + def render_system_prompt(self) -> str: prompt_context = self.build_prompt_context() parts = self.build_system_prompt_parts(prompt_context) diff --git a/src/run_state_store.py b/src/run_state_store.py index 860040a..5bc3635 100644 --- a/src/run_state_store.py +++ b/src/run_state_store.py @@ -230,6 +230,69 @@ class RunStateStore: return None return self._snapshot_from_row(conn, row) + def events_since( + self, + account_key: str, + session_id: str, + *, + after_event_seq: int = 0, + limit: int = 500, + ) -> dict[str, Any]: + with self._connect() as conn: + rows = conn.execute( + """ + select e.id, e.recorded_at, e.event_json, s.run_id, s.status + from run_events e + join run_states s on s.run_id = e.run_id + where s.account_key = ? + and s.session_id = ? + and e.id > ? + order by e.id asc + limit ? + """, + ( + account_key, + session_id, + max(0, int(after_event_seq)), + max(1, int(limit)), + ), + ).fetchall() + latest_row = conn.execute( + """ + select max(e.id) as latest_event_seq + from run_events e + join run_states s on s.run_id = e.run_id + where s.account_key = ? + and s.session_id = ? + """, + (account_key, session_id), + ).fetchone() + events: list[dict[str, Any]] = [] + for row in rows: + try: + event = json.loads(row['event_json']) + except (TypeError, json.JSONDecodeError): + continue + if not isinstance(event, dict): + continue + events.append( + { + 'event_seq': int(row['id']), + 'run_id': row['run_id'], + 'run_status': row['status'], + 'recorded_at': float(row['recorded_at'] or 0.0), + 'event': event, + } + ) + latest_event_seq = 0 + if latest_row is not None and latest_row['latest_event_seq'] is not None: + latest_event_seq = int(latest_row['latest_event_seq']) + return { + 'session_id': session_id, + 'latest_event_seq': latest_event_seq, + 'events': events, + } + def mark_interrupted_if_active( self, run_id: str, diff --git a/src/session_store.py b/src/session_store.py index 1c6242d..d553cba 100644 --- a/src/session_store.py +++ b/src/session_store.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import hashlib import sqlite3 import time from dataclasses import asdict, dataclass @@ -84,6 +85,12 @@ def save_agent_session(session: StoredAgentSession, directory: Path | None = Non path = session_dir / 'session.json' payload = asdict(session) _write_agent_session_db_payload(target_dir, session.session_id, payload, path) + _sync_agent_display_messages( + target_dir, + session.session_id, + tuple(message for message in session.display_messages if isinstance(message, dict)) + or tuple(message for message in session.messages if isinstance(message, dict)), + ) path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding='utf-8') return path @@ -131,6 +138,266 @@ def list_agent_sessions( return sorted(sessions.values(), key=lambda item: item[1], reverse=True) +def agent_session_delta( + session_id: str, + directory: Path | None = None, + *, + after_message_seq: int = 0, + limit: int = 500, +) -> dict[str, Any]: + target_dir = directory or DEFAULT_AGENT_SESSION_DIR + try: + stored = load_agent_session(session_id, directory=target_dir) + except (FileNotFoundError, json.JSONDecodeError, KeyError, TypeError, ValueError): + stored = None + if stored is not None: + _sync_agent_display_messages( + target_dir, + session_id, + stored.display_messages or stored.messages, + ) + with _connect_agent_session_db(target_dir) as conn: + latest_row = conn.execute( + """ + select max(seq) as latest_seq + from agent_display_messages + where session_id = ? + """, + (session_id,), + ).fetchone() + rows = conn.execute( + """ + select seq, message_json, updated_at + from agent_display_messages + where session_id = ? and seq > ? + order by seq asc + limit ? + """, + (session_id, max(0, int(after_message_seq)), max(1, int(limit))), + ).fetchall() + messages: list[dict[str, Any]] = [] + for row in rows: + try: + message = json.loads(row['message_json']) + except (TypeError, json.JSONDecodeError): + continue + if isinstance(message, dict): + messages.append( + { + 'seq': int(row['seq']), + 'updated_at': float(row['updated_at'] or 0.0), + 'message': message, + } + ) + latest_seq = 0 + if latest_row is not None and latest_row['latest_seq'] is not None: + latest_seq = int(latest_row['latest_seq']) + return { + 'session_id': session_id, + 'latest_message_seq': latest_seq, + 'messages': messages, + 'session': ( + { + 'session_id': stored.session_id, + 'turns': stored.turns, + 'tool_calls': stored.tool_calls, + 'usage': stored.usage, + 'total_cost_usd': stored.total_cost_usd, + 'model': stored.model_config.get('model'), + 'is_training': stored.is_training, + 'session_metadata': stored.session_metadata or {}, + } + if stored is not None + else {'session_id': session_id} + ), + } + + +def enqueue_session_input( + session_id: str, + *, + content: str, + directory: Path | None = None, + run_id: str | None = None, + kind: str = 'next_turn', + status: str = 'pending', + metadata: dict[str, Any] | None = None, +) -> dict[str, Any]: + target_dir = directory or DEFAULT_AGENT_SESSION_DIR + now = time.time() + normalized_kind = kind if kind in {'next_turn', 'guidance'} else 'next_turn' + with _connect_agent_session_db(target_dir) as conn: + cursor = conn.execute( + """ + insert into agent_input_queue ( + session_id, run_id, kind, status, content, + created_at, updated_at, metadata_json + ) + values (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + session_id, + run_id or '', + normalized_kind, + status, + content, + now, + now, + json.dumps(metadata or {}, ensure_ascii=False, sort_keys=True), + ), + ) + item_id = int(cursor.lastrowid) + return { + 'id': item_id, + 'session_id': session_id, + 'run_id': run_id or '', + 'kind': normalized_kind, + 'status': status, + 'content': content, + 'created_at': now, + 'updated_at': now, + 'metadata': metadata or {}, + } + + +def list_session_input_queue( + session_id: str, + directory: Path | None = None, + *, + statuses: tuple[str, ...] = ('pending',), + limit: int = 100, +) -> list[dict[str, Any]]: + target_dir = directory or DEFAULT_AGENT_SESSION_DIR + placeholders = ','.join('?' for _ in statuses) + with _connect_agent_session_db(target_dir) as conn: + rows = conn.execute( + f""" + select * + from agent_input_queue + where session_id = ? + and status in ({placeholders}) + order by id asc + limit ? + """, + (session_id, *statuses, max(1, int(limit))), + ).fetchall() + return [_queue_row_to_dict(row) for row in rows] + + +def update_session_input_queue_item( + session_id: str, + item_id: int, + directory: Path | None = None, + *, + content: str | None = None, + kind: str | None = None, + status: str | None = None, + run_id: str | None = None, +) -> dict[str, Any] | None: + target_dir = directory or DEFAULT_AGENT_SESSION_DIR + assignments: list[str] = ['updated_at = ?'] + values: list[Any] = [time.time()] + if content is not None: + assignments.append('content = ?') + values.append(content) + if kind is not None: + assignments.append('kind = ?') + values.append(kind if kind in {'next_turn', 'guidance'} else 'next_turn') + if status is not None: + assignments.append('status = ?') + values.append(status) + if status in {'consumed', 'cancelled'}: + assignments.append('consumed_at = coalesce(consumed_at, ?)') + values.append(time.time()) + if run_id is not None: + assignments.append('run_id = ?') + values.append(run_id) + values.extend([session_id, int(item_id)]) + with _connect_agent_session_db(target_dir) as conn: + conn.execute( + f""" + update agent_input_queue + set {', '.join(assignments)} + where session_id = ? and id = ? + """, + values, + ) + row = conn.execute( + """ + select * + from agent_input_queue + where session_id = ? and id = ? + """, + (session_id, int(item_id)), + ).fetchone() + return _queue_row_to_dict(row) if row is not None else None + + +def cancel_session_input_queue( + session_id: str, + directory: Path | None = None, + *, + run_id: str | None = None, +) -> int: + target_dir = directory or DEFAULT_AGENT_SESSION_DIR + now = time.time() + where = 'session_id = ? and status = ?' + values: list[Any] = [session_id, 'pending'] + if run_id: + where += ' and (run_id = ? or run_id = ?)' + values.extend([run_id, '']) + with _connect_agent_session_db(target_dir) as conn: + cursor = conn.execute( + f""" + update agent_input_queue + set status = 'cancelled', + updated_at = ?, + consumed_at = coalesce(consumed_at, ?) + where {where} + """, + (now, now, *values), + ) + return int(cursor.rowcount or 0) + + +def consume_session_guidance( + session_id: str, + run_id: str, + directory: Path | None = None, + *, + limit: int = 10, +) -> list[dict[str, Any]]: + target_dir = directory or DEFAULT_AGENT_SESSION_DIR + now = time.time() + with _connect_agent_session_db(target_dir) as conn: + rows = conn.execute( + """ + select * + from agent_input_queue + where session_id = ? + and status = 'pending' + and kind = 'guidance' + and (run_id = ? or run_id = '') + order by id asc + limit ? + """, + (session_id, run_id, max(1, int(limit))), + ).fetchall() + ids = [int(row['id']) for row in rows] + if ids: + conn.execute( + f""" + update agent_input_queue + set status = 'consumed', + updated_at = ?, + consumed_at = ? + where id in ({','.join('?' for _ in ids)}) + """, + (now, now, *ids), + ) + return [_queue_row_to_dict(row) for row in rows] + + def delete_agent_session(session_id: str, directory: Path | None = None) -> bool: target_dir = directory or DEFAULT_AGENT_SESSION_DIR deleted = False @@ -140,6 +407,14 @@ def delete_agent_session(session_id: str, directory: Path | None = None) -> bool (session_id,), ) deleted = cursor.rowcount > 0 + conn.execute( + 'delete from agent_display_messages where session_id = ?', + (session_id,), + ) + conn.execute( + 'delete from agent_input_queue where session_id = ?', + (session_id,), + ) nested = target_dir / session_id if nested.exists(): import shutil @@ -288,6 +563,195 @@ def _write_agent_session_revision( ) +def _sync_agent_display_messages( + directory: Path, + session_id: str, + display_messages: tuple[JSONDict, ...], +) -> None: + if not display_messages: + return + now = time.time() + with _connect_agent_session_db(directory) as conn: + existing_rows = conn.execute( + """ + select seq, message_key, message_json + from agent_display_messages + where session_id = ? + """, + (session_id,), + ).fetchall() + existing: dict[int, sqlite3.Row] = { + int(row['seq']): row for row in existing_rows + } + existing_by_key: dict[str, sqlite3.Row] = { + str(row['message_key'] or ''): row + for row in existing_rows + if str(row['message_key'] or '') + } + next_seq = max(existing.keys(), default=0) + 1 + for index, message in enumerate(display_messages, start=1): + if not isinstance(message, dict): + continue + message_json = json.dumps(message, ensure_ascii=False, sort_keys=True) + message_key = _display_message_key(message) + row = existing.get(index) + keyed_row = existing_by_key.get(message_key) + if keyed_row is not None and keyed_row['message_json'] == message_json: + continue + if keyed_row is not None: + try: + keyed_old_message = json.loads(keyed_row['message_json']) + except (TypeError, json.JSONDecodeError): + keyed_old_message = {} + if not isinstance(keyed_old_message, dict): + keyed_old_message = {} + if _is_replaceable_display_message(keyed_old_message): + conn.execute( + """ + update agent_display_messages + set role = ?, + updated_at = ?, + message_json = ? + where session_id = ? and seq = ? + """, + ( + str(message.get('role') or ''), + now, + message_json, + session_id, + int(keyed_row['seq']), + ), + ) + continue + if row is None: + conn.execute( + """ + insert into agent_display_messages ( + session_id, seq, message_key, role, updated_at, message_json + ) + values (?, ?, ?, ?, ?, ?) + """, + ( + session_id, + index, + message_key, + str(message.get('role') or ''), + now, + message_json, + ), + ) + continue + if row['message_json'] == message_json: + continue + try: + old_message = json.loads(row['message_json']) + except (TypeError, json.JSONDecodeError): + old_message = {} + if not isinstance(old_message, dict): + old_message = {} + old_key = str(row['message_key'] or '') + should_update = ( + old_key == message_key + or _is_replaceable_display_message(old_message) + or _same_role_and_content(old_message, message) + ) + if not should_update: + append_seq = next_seq + next_seq += 1 + conn.execute( + """ + insert into agent_display_messages ( + session_id, seq, message_key, role, updated_at, message_json + ) + values (?, ?, ?, ?, ?, ?) + """, + ( + session_id, + append_seq, + message_key, + str(message.get('role') or ''), + now, + message_json, + ), + ) + existing_by_key[message_key] = { + 'seq': append_seq, + 'message_key': message_key, + 'message_json': message_json, + } # type: ignore[assignment] + continue + conn.execute( + """ + update agent_display_messages + set message_key = ?, + role = ?, + updated_at = ?, + message_json = ? + where session_id = ? and seq = ? + """, + ( + message_key, + str(message.get('role') or ''), + now, + message_json, + session_id, + index, + ), + ) + + +def _display_message_key(message: JSONDict) -> str: + message_id = message.get('message_id') + if isinstance(message_id, str) and message_id: + return f'id:{message_id}' + tool_call_id = message.get('tool_call_id') + if isinstance(tool_call_id, str) and tool_call_id: + return f'tool:{tool_call_id}' + role = str(message.get('role') or '') + content = '' if message.get('content') is None else str(message.get('content') or '') + digest = hashlib.sha1(f'{role}\0{content}'.encode('utf-8')).hexdigest() + return f'hash:{digest}' + + +def _is_replaceable_display_message(message: JSONDict) -> bool: + metadata = message.get('metadata') + if not isinstance(metadata, dict): + return False + if metadata.get('placeholder') is True: + return True + return metadata.get('kind') in {'run_status'} + + +def _same_role_and_content(left: JSONDict, right: JSONDict) -> bool: + return ( + str(left.get('role') or '') == str(right.get('role') or '') + and str(left.get('content') or '') == str(right.get('content') or '') + ) + + +def _queue_row_to_dict(row: sqlite3.Row) -> dict[str, Any]: + try: + metadata = json.loads(row['metadata_json'] or '{}') + except (TypeError, json.JSONDecodeError): + metadata = {} + return { + 'id': int(row['id']), + 'session_id': str(row['session_id'] or ''), + 'run_id': str(row['run_id'] or ''), + 'kind': str(row['kind'] or ''), + 'status': str(row['status'] or ''), + 'content': str(row['content'] or ''), + 'created_at': float(row['created_at'] or 0.0), + 'updated_at': float(row['updated_at'] or 0.0), + 'consumed_at': ( + float(row['consumed_at']) + if row['consumed_at'] is not None + else None + ), + 'metadata': metadata if isinstance(metadata, dict) else {}, + } + + def _read_agent_session_db_payload( directory: Path, session_id: str, @@ -365,6 +829,53 @@ def _connect_agent_session_db(directory: Path) -> sqlite3.Connection: on agent_session_revisions(session_id, revision_id) """ ) + conn.execute( + """ + create table if not exists agent_display_messages ( + session_id text not null, + seq integer not null, + message_key text not null, + role text default '', + updated_at real not null, + message_json text not null, + primary key(session_id, seq) + ) + """ + ) + conn.execute( + """ + create index if not exists idx_agent_display_messages_session + on agent_display_messages(session_id, seq) + """ + ) + conn.execute( + """ + create table if not exists agent_input_queue ( + id integer primary key autoincrement, + session_id text not null, + run_id text default '', + kind text not null, + status text not null, + content text not null, + created_at real not null, + updated_at real not null, + consumed_at real, + metadata_json text default '{}' + ) + """ + ) + conn.execute( + """ + create index if not exists idx_agent_input_queue_session + on agent_input_queue(session_id, status, id) + """ + ) + conn.execute( + """ + create index if not exists idx_agent_input_queue_run + on agent_input_queue(session_id, run_id, status, id) + """ + ) return conn