feat: non-blocking Agent and editable AI resume canvas

backend/app/routes/main_agent.py

@@ -963,12 +963,15 @@ async def run_and_record() -> dict[str, Any]:
             assistant_message = str(result.get("assistant_message") or "").strip()
             completed_conversation = conversation
             if assistant_message:
+                # Re-read: steering messages and parked sibling Runs may have
+                # appended to this conversation while the Run was executing.
+                latest_conversation = get_conversation(conversation["id"]) or conversation
                 completed_conversation = await _ui_operation_outputs(
                     "save_harness_conversation",
                     {
                         "conversation_id": conversation["id"],
                         "messages": [
-                            *conversation["messages"],
+                            *(latest_conversation.get("messages") or []),
                             {"role": "assistant", "content": assistant_message},
                         ],
                     },
@@ -1263,6 +1266,42 @@ async def resume_runtime_run(run_id: str) -> dict[str, Any]:
         raise HTTPException(status_code=400, detail=safe_error_message(exc))
 
 
+class RuntimeSteerBody(BaseModel):
+    message: str = Field(min_length=1, max_length=4000)
+
+
+@runtime_router.post("/runtime/runs/{run_id}/steer")
+async def steer_runtime_run(run_id: str, body: RuntimeSteerBody) -> dict[str, Any]:
+    """Deliver a mid-run user message to the live Run (D-1: input never locks).
+
+    ``not_running`` means the Run is not executing a prompt; the client then
+    sends the text as a normal follow-up turn. Nothing here approves anything.
+    """
+    try:
+        provider = await _provider_for_run(run_id)
+        steer = getattr(provider, "steer_run", None)
+        if steer is None:
+            return {"ok": True, "disposition": "not_running"}
+        result = await steer(run_id, message=body.message)
+    except ValueError as exc:
+        raise HTTPException(status_code=404, detail=safe_error_message(exc))
+    if result.get("disposition") == "steered":
+        conversation_id = str((result.get("run") or {}).get("conversation_id") or "")
+        conversation = get_conversation(conversation_id) if conversation_id else None
+        if conversation is not None:
+            await _ui_operation_outputs(
+                "save_harness_conversation",
+                {
+                    "conversation_id": conversation_id,
+                    "messages": [
+                        *(conversation.get("messages") or []),
+                        {"role": "user", "content": body.message},
+                    ],
+                },
+            )
+    return {"ok": True, "disposition": result.get("disposition") or "not_running"}
+
+
 @runtime_router.post("/runtime/runs/{run_id}/abort")
 async def abort_runtime_run(run_id: str) -> dict[str, Any]:
     try:

backend/app/services/agent_runtime.py

@@ -89,6 +89,7 @@
     "run.waiting_confirmation": "approval.requested",
     "input.required": "approval.requested",
     "input.answered": "approval.resolved",
+    "input.steered": "reasoning.status",
     "run.waiting_input": "approval.requested",
     "continuation.requested": "reasoning.status",
     "approval.requested": "approval.requested",
@@ -294,6 +295,11 @@ async def abort_run(self, run_id: str) -> dict[str, Any]:
 
         return await abort_embedded_agent_run(run_id)
 
+    async def steer_run(self, run_id: str, *, message: str) -> dict[str, Any]:
+        from app.services.embedded_agent_host import steer_embedded_agent_run
+
+        return await steer_embedded_agent_run(run_id, message)
+
 
 class ReplayAgentRunProvider:
     """Durable, deterministic Main Agent provider for local acceptance paths."""

backend/app/services/embedded_agent_host.py

@@ -45,6 +45,7 @@
 from app.services.embedded_agent_worker import (
     PROTOCOL_VERSION,
     EmbeddedAgentWorker,
+    EmbeddedAgentWorkerError,
     get_embedded_agent_worker,
 )
 from app.services.security_redaction import safe_error_message
@@ -991,6 +992,16 @@ async def _request_user_input(args: dict[str, Any]) -> dict[str, Any]:
             await publish("run.completed", {"status": "completed"})
             await active_worker.dispose_run(run_id)
             current = await load_agent_run(run_id) or current
+        if current["status"] in _PARKED_WAITING_STATUSES and active_worker.active_run_id == run_id:
+            # D-1: a Run that waits on the user must not hold the single
+            # embedded worker. Its session file stays on disk; approvals and
+            # answers already resume through continuations, which restart the
+            # worker from that file. Parking lets the user keep chatting.
+            try:
+                await active_worker.dispose_run(run_id)
+            except Exception:
+                pass
+            current = await load_agent_run(run_id) or current
         await persist_and_publish(
             "guardian.reviewed",
             {
@@ -1339,6 +1350,43 @@ async def reject_embedded_agent_action(
     return result
 
 
+_PARKED_WAITING_STATUSES = frozenset({"waiting_confirmation", "waiting_input", "waiting_decision"})
+
+
+async def steer_embedded_agent_run(
+    run_id: str,
+    message: str,
+    *,
+    worker: EmbeddedAgentWorker | None = None,
+) -> dict[str, Any]:
+    """Inject a user message into a live Run at its next tool boundary.
+
+    Steering is collaboration input, never authorization. When the Run is not
+    executing a prompt right now (finished, parked or waiting on the user) the
+    caller gets ``not_running`` and should start a follow-up turn instead.
+    """
+
+    text = str(message or "").strip()
+    if not text:
+        raise ValueError("引导消息不能为空。")
+    run = await load_agent_run(run_id)
+    if run is None:
+        raise ValueError(f"Agent Run {run_id} 不存在。")
+    active_worker = worker or get_embedded_agent_worker()
+    if active_worker.active_run_id != run_id or not active_worker.prompt_active:
+        return {"ok": True, "disposition": "not_running", "run": run}
+    try:
+        await active_worker.steer_run(run_id, text[:4000])
+    except EmbeddedAgentWorkerError:
+        return {"ok": True, "disposition": "not_running", "run": run}
+    await append_agent_run_event(
+        run_id,
+        event_type="input.steered",
+        payload={"characters": len(text)},
+    )
+    return {"ok": True, "disposition": "steered", "run": run}
+
+
 async def abort_embedded_agent_run(
     run_id: str,
     *,

backend/app/services/resume_workspace.py

@@ -614,6 +614,18 @@ def _find_section(resume: Resume, row: Optional[dict[str, Any]]) -> Optional[Res
     )
 
 
+def _change_conflicts_with_manual_edit(resume: Resume, diff: dict[str, Any]) -> bool:
+    """True when the section this change targets no longer matches the proposal's ``before``."""
+    change_type = str(diff.get("change_type") or "modified")
+    before = diff.get("before") if isinstance(diff.get("before"), dict) else None
+    if change_type == "added" or before is None:
+        return False
+    target = _find_section(resume, before)
+    if target is None:
+        return True
+    return _json_hash(target.content_json or []) != _json_hash(before.get("content_json") or [])
+
+
 async def review_resume_proposal_item(
     proposal_id: str, resume_id: int, change_id: str, action: str, edited_text: str = "",
 ) -> dict[str, Any]:
@@ -723,11 +735,19 @@ async def _review_resume_proposal_items(
                     await db.commit()
                     raise ValueError(proposal.review_note)
             if expected_hash and workspace_content_hash(resume) != expected_hash:
-                proposal.status = "stale"
-                proposal.review_note = "用户手动修改后，原提案已过期，请重新生成或重新计算"
-                proposal.reviewed_at = _now()
-                await db.commit()
-                raise ValueError(proposal.review_note)
+                # 人工修改优先，冲突按段落判断（docs/05 §5）：用户改了别处，
+                # 这条建议仍可采用；只有建议要改的那一段在生成后被用户改过，
+                # 才拒绝采用这一条，且绝不覆盖用户的版本。整份提案不因此失效。
+                conflicted = [
+                    change_id
+                    for change_id in pending_ids
+                    if _change_conflicts_with_manual_edit(resume, diffs[change_id])
+                ]
+                if conflicted:
+                    raise ValueError(
+                        "这一段在 AI 建议生成后被你改过，建议基于旧内容，不能直接采用；"
+                        "可以跳过，或让 AI 基于你的新版本重新建议"
+                    )
         for clean_change_id in pending_ids:
             diff = diffs[clean_change_id]
             if clean_action == "accept":

backend/tests/test_agent_never_locks.py

@@ -0,0 +1,100 @@
+"""docs/02 D-1 / D-2: a Run waiting on the user never holds the Agent hostage.
+
+* A Run that reaches ``waiting_confirmation`` parks: it releases the single
+  embedded worker so the user can start another turn, while its session file
+  and plan stay durable for the later approval continuation.
+* A mid-run user message is steered into the live Run, or reported as
+  ``not_running`` so the client sends it as a normal follow-up turn.
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+import unittest
+from pathlib import Path
+from typing import Any
+
+BACKEND_DIR = Path(__file__).resolve().parents[1]
+os.chdir(BACKEND_DIR)
+if str(BACKEND_DIR) not in sys.path:
+    sys.path.insert(0, str(BACKEND_DIR))
+TESTS_DIR = Path(__file__).resolve().parent
+if str(TESTS_DIR) not in sys.path:
+    sys.path.insert(0, str(TESTS_DIR))
+
+from app.database import init_db
+from app.services.agent_run_state import create_agent_run, list_agent_run_events, load_agent_run
+from app.services.embedded_agent_host import start_embedded_agent_run, steer_embedded_agent_run
+from test_embedded_agent_host import FakeEmbeddedWorker
+
+PROVIDER = {"name": "fixture", "model": "fixture"}
+METADATA = {"runtime": "python_agent", "provider_id": "embedded"}
+
+
+class SteerableWorker:
+    def __init__(self, run_id: str, *, prompt_active: bool) -> None:
+        self.active_run_id = run_id
+        self.prompt_active = prompt_active
+        self.steered: list[str] = []
+
+    async def steer_run(self, run_id: str, message: str) -> dict[str, Any]:
+        assert run_id == self.active_run_id
+        self.steered.append(message)
+        return {"run_id": run_id, "disposition": "queued"}
+
+
+class AgentNeverLocksTests(unittest.TestCase):
+    def test_waiting_confirmation_parks_the_worker_but_keeps_the_run(self) -> None:
+        worker = FakeEmbeddedWorker()
+
+        async def run() -> tuple[dict[str, Any], str | None, dict[str, Any] | None]:
+            await init_db()
+            from app.database import async_session
+            from app.models.models import Job
+            async with async_session() as db:
+                if await db.get(Job, 74291) is None:
+                    db.add(Job(id=74291, title="Synthetic recovery role", company="Fixture",
+                               raw_description="Public synthetic JD", hash_key="host-recovery-job"))
+                    await db.commit()
+            started = await start_embedded_agent_run(
+                message="帮我调研这个岗位",
+                skill_id="company_research",
+                conversation_id="never-locks-park",
+                worker=worker,
+                provider_config=PROVIDER,
+                provider_metadata=METADATA,
+            )
+            return started, worker.active_run_id, await load_agent_run(started["run"]["id"])
+
+        started, active_run_id, stored = asyncio.run(run())
+        self.assertTrue(started["ok"], started)
+        self.assertEqual(stored["status"], "waiting_confirmation")
+        self.assertIsNone(active_run_id)
+        self.assertTrue(started["run"]["final_result"]["requires_confirmation"])
+
+    def test_steer_reaches_only_a_live_prompt(self) -> None:
+        async def run() -> tuple[dict[str, Any], dict[str, Any], list[str], list[str]]:
+            await init_db()
+            record = await create_agent_run(conversation_id="never-locks-steer", goal="steer fixture", mode="skill", skill_id="discovery", actions=[])
+            run_id = record["id"]
+            live = SteerableWorker(run_id, prompt_active=True)
+            steered = await steer_embedded_agent_run(run_id, "  换成只看上海的岗位  ", worker=live)
+            idle = SteerableWorker(run_id, prompt_active=False)
+            not_running = await steer_embedded_agent_run(run_id, "这条应作为新一轮", worker=idle)
+            events = [event["type"] for event in await list_agent_run_events(run_id)]
+            return steered, not_running, live.steered + idle.steered, events
+
+        steered, not_running, delivered, events = asyncio.run(run())
+        self.assertEqual(steered["disposition"], "steered")
+        self.assertEqual(not_running["disposition"], "not_running")
+        self.assertEqual(delivered, ["换成只看上海的岗位"])
+        self.assertEqual(events.count("input.steered"), 1)
+
+    def test_empty_steer_is_rejected(self) -> None:
+        with self.assertRaises(ValueError):
+            asyncio.run(steer_embedded_agent_run("run_0000000000000000", "   "))
+
+
+if __name__ == "__main__":
+    unittest.main()