Restore pending user message during active replay

This commit is contained in:
wuyang6
2026-05-12 10:58:25 +08:00
parent 26fe0f6a9d
commit 1bd80db2db
2 changed files with 59 additions and 5 deletions
+9 -5
View File
@@ -354,6 +354,7 @@ class RunRecord:
updated_at: float updated_at: float
cancel_event: threading.Event cancel_event: threading.Event
process_registry: RunProcessRegistry process_registry: RunProcessRegistry
pending_prompt: str = ''
current_stage: str = '' current_stage: str = ''
error: str = '' error: str = ''
events: list[dict[str, Any]] = field(default_factory=list) events: list[dict[str, Any]] = field(default_factory=list)
@@ -365,7 +366,7 @@ class RunManager:
self._runs: dict[str, RunRecord] = {} self._runs: dict[str, RunRecord] = {}
self._latest_by_session: dict[tuple[str, str], str] = {} self._latest_by_session: dict[tuple[str, str], str] = {}
def start(self, account_key: str, session_id: str) -> RunRecord: def start(self, account_key: str, session_id: str, pending_prompt: str = '') -> RunRecord:
now = time.time() now = time.time()
record = RunRecord( record = RunRecord(
run_id=uuid4().hex, run_id=uuid4().hex,
@@ -376,6 +377,7 @@ class RunManager:
updated_at=now, updated_at=now,
cancel_event=threading.Event(), cancel_event=threading.Event(),
process_registry=RunProcessRegistry(), process_registry=RunProcessRegistry(),
pending_prompt=pending_prompt,
) )
with self._lock: with self._lock:
self._runs[record.run_id] = record self._runs[record.run_id] = record
@@ -442,6 +444,7 @@ class RunManager:
'current_stage': record.current_stage, 'current_stage': record.current_stage,
'started_at': record.started_at, 'started_at': record.started_at,
'updated_at': record.updated_at, 'updated_at': record.updated_at,
'pending_prompt': record.pending_prompt,
'error': record.error, 'error': record.error,
'events': [dict(event) for event in record.events], 'events': [dict(event) for event in record.events],
} }
@@ -1303,7 +1306,8 @@ def create_app(state: AgentState) -> FastAPI:
request.resume_session_id or request.session_id request.resume_session_id or request.session_id
) or uuid4().hex ) or uuid4().hex
account_key = state._account_key(request.account_id) account_key = state._account_key(request.account_id)
run_record = state.run_manager.start(account_key, requested_session_id) prompt = request.prompt.strip()
run_record = state.run_manager.start(account_key, requested_session_id, prompt)
run_lock = state.run_lock_for(request.account_id, requested_session_id) run_lock = state.run_lock_for(request.account_id, requested_session_id)
agent = state.agent_for(request.account_id, requested_session_id) agent = state.agent_for(request.account_id, requested_session_id)
config = state.config_for(request.account_id) config = state.config_for(request.account_id)
@@ -1326,7 +1330,7 @@ def create_app(state: AgentState) -> FastAPI:
directory=session_directory, directory=session_directory,
agent=agent, agent=agent,
session_id=requested_session_id, session_id=requested_session_id,
prompt=request.prompt.strip(), prompt=prompt,
) )
if run_lock.locked(): if run_lock.locked():
queued_event = { queued_event = {
@@ -1384,14 +1388,14 @@ def create_app(state: AgentState) -> FastAPI:
) )
stored = _sanitize_stored_session_for_resume(stored) stored = _sanitize_stored_session_for_resume(stored)
result = agent.resume( result = agent.resume(
request.prompt.strip(), prompt,
stored, stored,
runtime_context=request.runtime_context, runtime_context=request.runtime_context,
event_sink=emit_agent_event, event_sink=emit_agent_event,
) )
else: else:
result = agent.run( result = agent.run(
request.prompt.strip(), prompt,
session_id=requested_session_id, session_id=requested_session_id,
runtime_context=request.runtime_context, runtime_context=request.runtime_context,
event_sink=emit_agent_event, event_sink=emit_agent_event,
@@ -69,6 +69,7 @@ export type ClawRunStatus = {
| "interrupted"; | "interrupted";
current_stage?: string; current_stage?: string;
started_at?: number; started_at?: number;
pending_prompt?: string;
events?: ClawRunEvent[]; events?: ClawRunEvent[];
}; };
@@ -502,6 +503,32 @@ export function toReplayRepository(
previousId = id; previousId = id;
} }
if (isActiveRunStatus(runStatus)) { if (isActiveRunStatus(runStatus)) {
const pendingPrompt = resolvePendingPrompt(replayMessages, runStatus);
if (pendingPrompt) {
const id = `${fallbackSessionId}-pending-${runStatus.run_id ?? "active"}`;
const parts = [
{ type: "text", text: pendingPrompt },
] as UIMessage["parts"];
const uiMessage: UIMessage = {
id,
role: "user",
parts,
metadata: { sessionId: session.session_id ?? fallbackSessionId },
} as UIMessage;
const threadMessage = {
id,
role: "user",
createdAt: new Date(),
content: toThreadMessageContent(parts),
attachments: [],
metadata: {
custom: { sessionId: session.session_id ?? fallbackSessionId },
},
} as ThreadMessage;
bindExternalStoreMessage(threadMessage, uiMessage);
messages.push({ message: threadMessage, parentId: previousId });
previousId = id;
}
const id = `${fallbackSessionId}-run-${runStatus.run_id ?? "active"}`; const id = `${fallbackSessionId}-run-${runStatus.run_id ?? "active"}`;
const runStartedAtMs = resolveRunStartedAtMs(runStatus); const runStartedAtMs = resolveRunStartedAtMs(runStatus);
const parts = buildActiveRunParts(runStatus) as UIMessage["parts"]; const parts = buildActiveRunParts(runStatus) as UIMessage["parts"];
@@ -717,6 +744,29 @@ function isRunStatusMessage(message?: ClawStoredMessage) {
return message?.metadata?.kind === "run_status"; return message?.metadata?.kind === "run_status";
} }
function resolvePendingPrompt(
messages: readonly ClawStoredMessage[],
runStatus: ClawRunStatus,
) {
const pendingPrompt =
typeof runStatus.pending_prompt === "string"
? runStatus.pending_prompt.trim()
: "";
if (!pendingPrompt) return null;
for (let index = messages.length - 1; index >= 0; index -= 1) {
const message = messages[index];
if (!message || message.role === "system" || message.role === "tool")
continue;
const content = cleanStoredContent(message.content ?? "").trim();
if (!content || content.trimStart().startsWith("<system-reminder>"))
continue;
return message.role === "user" && content === pendingPrompt
? null
: pendingPrompt;
}
return pendingPrompt;
}
function normalizeRunTimestampMs(value: unknown) { function normalizeRunTimestampMs(value: unknown) {
if (typeof value !== "number" || !Number.isFinite(value)) return undefined; if (typeof value !== "number" || !Number.isFinite(value)) return undefined;
return value < 10_000_000_000 ? Math.round(value * 1000) : Math.round(value); return value < 10_000_000_000 ? Math.round(value * 1000) : Math.round(value);