From db296d2816ea1898ac5d98f0e73b929b2b6f68a3 Mon Sep 17 00:00:00 2001 From: jiayuqi7813 <1783671926@qq.com> Date: Sat, 22 Aug 2026 21:08:27 +0800 Subject: [PATCH 1/2] fix(web): repair worker setup and post-solve follow-ups --- apps/web/drivers.py | 163 +++++++++--- apps/web/run_manager.py | 240 ++++++++++++++---- apps/web/server.py | 75 ++++-- apps/web/ui/components/Conversation.tsx | 103 +++++--- .../web/ui/components/WorkerOrchestration.tsx | 46 ++-- apps/web/ui/lib/events.ts | 82 ++++++ apps/web/ui/lib/i18n.tsx | 7 +- apps/web/worker_config.py | 58 ++++- muteki/core/events.py | 3 + muteki/solver/btw.py | 3 +- muteki/solver/cli_driver.py | 38 ++- muteki/solver/cli_solver.py | 65 ++++- muteki/solver/container_exec.py | 9 +- muteki/solver/identity_model.py | 3 + muteki/solver/types.py | 5 +- muteki/swarm/coordinator_flags.py | 16 +- muteki/swarm/coordinator_loop.py | 31 ++- muteki/swarm/coordinator_race.py | 14 +- muteki/swarm/swarm.py | 3 +- tests/test_cli_executor.py | 44 +++- tests/test_standby_hitl.py | 186 ++++++++++++-- tests/test_web_control.py | 14 + tests/test_web_deck_ux.py | 53 ++++ tests/test_worker_config.py | 36 +++ 24 files changed, 1057 insertions(+), 240 deletions(-) diff --git a/apps/web/drivers.py b/apps/web/drivers.py index b5dbbe5..cd3226d 100644 --- a/apps/web/drivers.py +++ b/apps/web/drivers.py @@ -1079,6 +1079,14 @@ async def drive(run: Run) -> None: control_state_provider=control_state_provider, protocol2_session=protocol2_session, ) + if mgr is not None: + # Swarm owns the trusted winning outcome; RunManager owns storage that + # Worker containers cannot modify. Keep this hook process-local so + # experimental Swarm classes do not need a constructor API change. + swarm._winner_continuation_writer = ( # type: ignore[attr-defined] + lambda payload: mgr.persist_winner_continuation( + run.run_id, payload) + ) deferred_cleanup = False try: out = await swarm.run() @@ -1183,14 +1191,17 @@ async def _settle_incomplete_runtime() -> None: # ---- standby (post-solve HITL) ---------------------------------------------- # After a run finishes (or the server restarted), a human follow-up no longer has # a live swarm to reach. The standby driver COLD-STARTS a single worker from disk: -# it reads winner.json (the winning worker's CLI session) + the persisted -# shared_graph, resumes that SAME session, and serves one command — answer a +# it reads coordinator-owned continuation state + the persisted shared_graph, +# resumes that SAME session, and serves one command — answer a # question, mark the flag a false-positive and keep solving, or write a writeup. # Everything it needs is durable, so this works identically before and after a -# server restart. No winner.json (old run) → degrade to a fresh worker seeded with -# the board context. +# server restart. Older runs without private continuation metadata recover only +# non-sensitive identity from durable events and start a fresh session if needed. -def _standby_profile_for(engine: str, worker_profiles: list[dict[str, Any]]) -> dict[str, Any] | None: +def _standby_profile_for( + engine: str, + worker_profiles: list[dict[str, Any]], +) -> dict[str, Any] | None: """Pick the profile that should serve a post-solve standby command.""" if not worker_profiles: return None @@ -1350,6 +1361,7 @@ async def drive(run: Run) -> None: for key in ( "action", "target", "command_id", "request_id", "standing", "preempt_policy", "preemption", "flag", + "followup_id", ) if key in safe_cmd } @@ -1367,34 +1379,70 @@ async def drive(run: Run) -> None: return # no workspace → nothing durable to resume from graph_dir = root / "graph" - winner_path = root / "winner.json" arts = ArtifactStore(root=root / "arts") worker_root = root / "workers" worker_root.mkdir(parents=True, exist_ok=True) - winner: dict[str, Any] = {} - if winner_path.exists(): - try: - winner = json.loads(winner_path.read_text()) - except Exception: - winner = {} + winner = mgr.load_winner_continuation(run.run_id) - # Rebuild the Challenge: prefer the snapshot stored in winner.json. Older - # runs may not have winner.json, so recover the original launch payload from - # the durable JSONL before degrading to rail metadata. + # Rebuild the Challenge from coordinator-owned state. Older runs recover + # the launch payload and winning Worker identity from durable server events. + # Worker-writable workspace files never select profiles, credentials, + # backends, sessions or host paths. ch = winner.get("challenge") or {} - if not ch: + legacy_workers: dict[str, dict[str, str]] = {} + legacy_winner_actor = "" + if not ch or not winner.get("engine") or not winner.get("profile_id"): try: from muteki.core.events import EventType async for ev in run.store.replay(run.run_id): - if ev.event_type in { + payload = ev.payload or {} + if not ch and ev.event_type in { EventType.RUN_PREPARING, EventType.RUN_STARTED, }: - ch = (ev.payload or {}).get("challenge") or {} - if ch: - break + ch = payload.get("challenge") or {} + + solver_id = str(ev.solver_id or "").strip() + if solver_id and ev.event_type in { + EventType.WORKER_STATUS, EventType.WORKER_LIFECYCLE, + }: + current = legacy_workers.setdefault(solver_id, {}) + for source, target in ( + ("engine", "engine"), + ("profile_id", "profile_id"), + ("session", "session"), + ): + value = str(payload.get(source) or "").strip() + if value: + current[target] = value + + kind = str(payload.get("kind") or "") + actor = "" + if (ev.event_type is EventType.BLACKBOARD_DELTA + and kind == "flag_found"): + actor = str(payload.get("actor") or solver_id).strip() + elif (ev.event_type is EventType.SOLVE_GRAPH_DELTA + and kind == "flag"): + actor = solver_id + elif (ev.event_type is EventType.INSIGHT_BUS_EVENT + and kind == "FlagFound"): + actor = str(payload.get("by") or solver_id).strip() + elif ev.event_type is EventType.FLAG_ACCEPTED: + actor = str(payload.get("actor") or solver_id).strip() + if actor and actor != "coordinator": + legacy_winner_actor = actor + except Exception: - ch = {} + pass + if legacy_winner_actor: + legacy_worker = legacy_workers.get(legacy_winner_actor) or {} + winner.setdefault("worker_id", legacy_winner_actor) + if legacy_worker.get("engine"): + winner.setdefault("engine", legacy_worker["engine"]) + if legacy_worker.get("profile_id"): + winner.setdefault("profile_id", legacy_worker["profile_id"]) + if legacy_worker.get("session"): + winner.setdefault("session", legacy_worker["session"]) mode = ch.get("mode") or "ctf" if mode not in ("ctf", "pentest"): mode = "ctf" @@ -1419,9 +1467,7 @@ async def drive(run: Run) -> None: flag_format=ch.get("flag_format", _DEFAULT_BRACE_FLAG_FORMAT), flag_format_hint=ch.get("flag_format_hint", ""), flag_format_wrapper=ch.get("flag_format_wrapper", ""), - # carry the run's flag mode across a post-solve standby re-solve so a - # mark_false/resolve doesn't silently revert a collection run to single - # flag (review #15). winner.json persists these in the challenge block. + # Carry the run's flag mode across a post-solve standby re-solve. expected_flags=int(ch.get("expected_flags") or 1), multi_flag=bool(ch.get("multi_flag", False)), verifier_rate_limited=bool(ch.get("verifier_rate_limited", False)), @@ -1436,7 +1482,14 @@ async def drive(run: Run) -> None: worker_profiles = wc.get("worker_profiles") or [] worker_network = str(wc.get("worker_network") or "bridge") winner_engine = str(winner.get("engine") or "claude") - profile = _standby_profile_for(winner_engine, worker_profiles) + winner_profile_ref = str( + winner.get("profile_id") or winner_engine + ).strip() + profile = _standby_profile_for(winner_profile_ref, worker_profiles) + if winner.get("profile_id") and profile is None: + raise RuntimeError( + "winning Worker profile is unavailable in current configuration" + ) transport = base_engine_for_profile(profile or winner_engine) worker_backend = resolve_worker_backend( request_backend=None, @@ -1609,7 +1662,7 @@ def _flag_from_operator_cmd() -> str: return raw if " " not in raw and len(raw) <= 240 else "" flag = (_flag_from_operator_cmd() if action == "mark_false" else "") or stored_flag - # multi-flag: the flags already collected (from winner.json), minus the one + # Multi-flag: the flags already collected, minus the one # the operator is marking false — so a mark_false re-solve worker is seeded # with the SURVIVING flags and re-finds only the missing one, not the rest. prior_flags = list( @@ -1651,8 +1704,17 @@ async def _emit_bb(kind: str, **fields: Any) -> None: except Exception: pass - workdir = str(winner.get("workdir") or "") - if not workdir or not Path(workdir).exists(): + workdir = "" + workdir_rel = str(winner.get("workdir_rel") or "").strip() + if workdir_rel: + try: + candidate = (root / workdir_rel).resolve() + candidate.relative_to(worker_root.resolve()) + if candidate.exists(): + workdir = str(candidate) + except (OSError, ValueError): + workdir = "" + if not workdir: workdir = str(worker_root / f"standby-{transport}") Path(workdir).mkdir(parents=True, exist_ok=True) if container is not None: @@ -1816,25 +1878,54 @@ async def _reap_runtime_until_exit() -> None: out = await worker.run() # writeup: persist the body to sessions/{id}/writeup.md (and it already # streamed to the chat as the worker's reply). + artifact_path = "" if action == "writeup" and getattr(out, "reply", ""): try: - (root / "writeup.md").write_text(out.reply) - except Exception: - pass - # mark_false that re-solved: refresh winner.json + run flags. Multi-flag: - # merge the re-found flag(s) into the run set (the invalidated one was - # already removed via reopen_after_false_positive) and persist the full - # list, mirroring Swarm._persist_winner. + writeup_path = root / "writeup.md" + writeup_path.write_text(out.reply) + artifact_path = str(writeup_path) + except Exception as exc: + raise RuntimeError("writeup artifact could not be persisted") from exc + if action in {"ask", "writeup"}: + from muteki.core.events import Event, EventType + await run.bus.emit(Event( + event_type=EventType.FOLLOWUP_COMPLETED, + run_id=run.run_id, + solver_id=solver_label, + payload={ + "followup_id": runtime_cmd.get("followup_id") or "", + "kind": action, + "text": getattr(out, "reply", "") or "", + "artifact_path": artifact_path, + }, + )) + # A successful false-positive re-solve becomes the next trusted + # continuation owner. Persist identifiers in coordinator-only storage; + # the workspace JSON remains a compatibility artifact. if action == "mark_false" and out.solved and out.flag: refound = list(getattr(out, "flags", None) or [out.flag]) run.merge_flags(refound) try: - (root / "winner.json").write_text(json.dumps({ + persisted = { "engine": out.engine, "worker_id": solver_label, "session": out.session, "workdir": out.workdir, "flag": run.flag, "flags": list(run.flags), "challenge": challenge.model_dump(), + "profile_id": str( + (profile or {}).get("id") + or (profile or {}).get("name") + or "" + ), + "backend": backend, + } + mgr.persist_winner_continuation(run.run_id, persisted) + (root / "winner.json").write_text(json.dumps({ + key: persisted[key] + for key in ( + "engine", "worker_id", "session", "workdir", + "flag", "flags", "challenge", "profile_id", + ) }, ensure_ascii=False, indent=2)) except Exception: pass diff --git a/apps/web/run_manager.py b/apps/web/run_manager.py index e4c179e..71f5b52 100644 --- a/apps/web/run_manager.py +++ b/apps/web/run_manager.py @@ -22,6 +22,7 @@ import shutil import stat import time +import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Any, Awaitable, Callable, Optional @@ -187,6 +188,9 @@ class Run: # Lifecycle admission state. Old-generation events are dropped before they # reach the durable log, and each generation may publish RUN_FINISHED once. terminal_generations: set[int] = field(default_factory=set) + # Ask/Writeup run after RUN_FINISHED but still publish durable, typed output. + # IDs stay active until their terminal follow-up event has been emitted. + active_followups: set[str] = field(default_factory=set) termination_reasons: dict[int, str] = field(default_factory=dict) # One task owns one readiness result for each exact participating profile # configuration. A continuation generation reuses the result; changing the @@ -451,8 +455,19 @@ async def _generation_filter(ev: Event) -> bool: if generation < run.execution_generation: return False payload.setdefault("control_generation", run.control_generation) + followup_types = { + EventType.FOLLOWUP_STARTED, + EventType.FOLLOWUP_COMPLETED, + EventType.FOLLOWUP_FAILED, + } + # These event types are exclusively produced by the post-run driver. + # Their own lifecycle ID provides UI correlation; admission must not + # depend on a transient in-memory set because a very short worker can + # complete while its control receipt is still settling. + allowed_followup = ev.event_type in followup_types if (generation in run.terminal_generations - and ev.event_type is not EventType.CONTROL_COMMAND): + and ev.event_type is not EventType.CONTROL_COMMAND + and not allowed_followup): # The terminal event closes this execution generation. A worker # subprocess may still flush a buffered frame while cancellation is # propagating, but that frame belongs to a closed runtime and must not @@ -547,6 +562,103 @@ def persist_profile_readiness(self, run: Run) -> None: os.chmod(temporary, 0o600) os.replace(temporary, path) + def _winner_continuation_path(self, run_id: str) -> Path: + return ( + self.control_root / self._safe_run_id(run_id) + / "winner-continuation.json" + ) + + def persist_winner_continuation( + self, run_id: str, payload: dict[str, Any], + ) -> None: + """Persist the resumable winner outside the Worker-writable workspace. + + Only stable identifiers and continuation data are retained. The current + Worker configuration remains authoritative for credentials, endpoints, + models and backend selection when a follow-up starts. + """ + directory = self.coordinator_control_dir(run_id) + path = directory / "winner-continuation.json" + temporary = directory / ( + f".winner-continuation-{os.getpid()}-{time.time_ns()}.tmp" + ) + workspace = self.workspace_dir(run_id).resolve() + worker_root = (workspace / "workers").resolve() + workdir_rel = "" + raw_workdir = str(payload.get("workdir") or "").strip() + if raw_workdir: + try: + resolved_workdir = Path(raw_workdir).resolve() + if self._is_within(resolved_workdir, worker_root): + workdir_rel = str(resolved_workdir.relative_to(workspace)) + except (OSError, ValueError): + workdir_rel = "" + + raw_profile = payload.get("profile") + profile_id = str(payload.get("profile_id") or "").strip() + if not profile_id and isinstance(raw_profile, dict): + profile_id = str( + raw_profile.get("id") or raw_profile.get("name") or "" + ).strip() + backend = str(payload.get("backend") or "").strip() + if backend not in {"local", "container"}: + backend = "" + flags = [ + str(value).strip() for value in list(payload.get("flags") or []) + if str(value).strip() + ] + first_flag = str(payload.get("flag") or "").strip() + if first_flag and first_flag not in flags: + flags.insert(0, first_flag) + challenge = payload.get("challenge") + stored = { + "version": 1, + "worker_id": str(payload.get("worker_id") or "").strip(), + "profile_id": profile_id, + "engine": str(payload.get("engine") or "").strip(), + "session": str(payload.get("session") or "").strip(), + "workdir_rel": workdir_rel, + "backend": backend, + "flag": flags[0] if flags else "", + "flags": list(dict.fromkeys(flags)), + "challenge": challenge if isinstance(challenge, dict) else {}, + } + temporary.write_text( + json.dumps(stored, ensure_ascii=False, sort_keys=True), + encoding="utf-8", + ) + os.chmod(temporary, 0o600) + os.replace(temporary, path) + + def load_winner_continuation(self, run_id: str) -> dict[str, Any]: + """Load coordinator-owned continuation metadata, failing closed.""" + path = self._winner_continuation_path(run_id) + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (FileNotFoundError, OSError, ValueError, TypeError): + return {} + if not isinstance(payload, dict) or payload.get("version") != 1: + return {} + return dict(payload) + + def update_winner_continuation_flags( + self, run_id: str, flags: list[str], + ) -> None: + continuation = self.load_winner_continuation(run_id) + if not continuation: + return + continuation["flags"] = list(dict.fromkeys( + str(value).strip() for value in flags if str(value).strip() + )) + continuation["flag"] = ( + continuation["flags"][0] if continuation["flags"] else "" + ) + rel = str(continuation.pop("workdir_rel", "") or "") + continuation["workdir"] = ( + str((self.workspace_dir(run_id) / rel).resolve()) if rel else "" + ) + self.persist_winner_continuation(run_id, continuation) + def _apply_meta(self, run: "Run") -> None: """Overlay persisted operator metadata (pin/archive/rename) onto a run.""" m = self.meta.get(run.run_id) @@ -1696,27 +1808,22 @@ def _standby_scope_matches_winner(self, run: Run, target: str) -> bool: return True if scope.kind.value in {"run", "challenge"}: return scope.value == run.run_id - try: - import json - winner_path = self.workspace_dir(run.run_id) / "winner.json" - winner = json.loads(winner_path.read_text()) - except Exception: + winner = self.load_winner_continuation(run.run_id) + if not winner: return False if scope.kind.value == "worker": persisted_worker = str(winner.get("worker_id") or "") return bool(persisted_worker and persisted_worker == scope.value) if scope.kind.value == "engine": return str(winner.get("engine") or "") == scope.value - # Intent/lane identity is not persisted in winner.json; never widen it to + # Intent/lane identity is not persisted in continuation state; never widen it to # the winner merely because that is the only standby session available. return False def _register_standby_winner(self, run: Run) -> None: """Project the persisted winner as the only valid finished-run mailbox.""" try: - import json - winner = json.loads( - (self.workspace_dir(run.run_id) / "winner.json").read_text()) + winner = self.load_winner_continuation(run.run_id) worker_id = str(winner.get("worker_id") or "").strip() if not worker_id: return @@ -2074,6 +2181,13 @@ async def _standby_control(wire: dict[str, Any]) -> Any: busy = self._standby_busy(run) target = str(wire.get("target") or "global") exact_text = str(wire.get("text") or wire.get("hint") or "").strip() + if action == "ask" and not exact_text: + return { + "state": "unknown", + "detail": "ask requires a question", + "target_ids": [], + "metadata": {"code": "followup_question_required"}, + } if action in self._OFFLINE_CONTROL_ACTIONS: # The actor already expired typed ContextResources. Atomically @@ -2228,33 +2342,24 @@ async def _standby_control(wire: dict[str, Any]) -> Any: graph.close() run.invalidate_flag(flag) - # Keep the post-solve snapshot aligned immediately. A failed - # re-solve must not let a later writeup/ask seed workers from a - # winner.json that still contains the invalidated flag. try: - import json - winner_path = self.workspace_dir(run.run_id) / "winner.json" - winner = (json.loads(winner_path.read_text()) - if winner_path.exists() else {}) - surviving = [ - value for value in list(winner.get("flags") or run.flags) - if value and value != flag - ] - winner["flags"] = surviving - winner["flag"] = surviving[0] if surviving else "" - temp = winner_path.with_suffix(".json.tmp") - temp.write_text(json.dumps( - winner, ensure_ascii=False, indent=2)) - os.replace(temp, winner_path) + self.update_winner_continuation_flags( + run.run_id, list(run.flags)) except Exception: - # The graph/run projection remains authoritative; standby also - # reads graph flags first. Snapshot repair is retried naturally - # on the next successful solve. pass runtime_wire = dict(wire) runtime_wire["_control_mark_false_applied"] = True try: from muteki.core.events import blackboard_delta_payload + # A false-positive invalidation resumes solving and therefore + # owns a new execution generation. This lets its normal + # RUN_STARTED/RUN_FINISHED stream through while the completed + # generation remains sealed against late worker frames. + run.execution_generation += 1 + self._fresh_bus(run) + run.finished = False + run.solved = False + run.paused = False await run.bus.emit(Event( event_type=EventType.BLACKBOARD_DELTA, run_id=run.run_id, @@ -2264,7 +2369,10 @@ async def _standby_control(wire: dict[str, Any]) -> Any: await run.bus.emit(Event( event_type=EventType.RUN_REOPENED, run_id=run.run_id, - payload={"flag": flag}, + payload={ + "flag": flag, + "execution_generation": run.execution_generation, + }, )) except Exception: pass @@ -2904,8 +3012,9 @@ async def _resolve_launching( so the persisted shared_graph (verified facts / dead-ends) carries straight over — the swarm builds ON the prior evidence instead of from scratch. - The challenge is reconstructed from winner.json (the durable run snapshot), - falling back to the run's rail metadata. Caller-supplied `body` fields win + The challenge is reconstructed from coordinator-owned continuation state, + falling back to durable lifecycle events and rail metadata. Caller-supplied + `body` fields win (e.g. an operator hint folded into the description, a new target).""" if run.runtime_incomplete and not await self._settle_incomplete_runtime( run, timeout=self._standby_cancel_timeout()): @@ -2929,15 +3038,8 @@ async def _resolve_launching( if not await self._drain_control_before_launch(run_id, run): return False - # rebuild the challenge body from the durable winner.json snapshot. - ch: dict[str, Any] = {} - try: - import json - wp = self.workspace_dir(run_id) / "winner.json" - if wp.exists(): - ch = (json.loads(wp.read_text()) or {}).get("challenge") or {} - except Exception: - ch = {} + continuation = self.load_winner_continuation(run_id) + ch = continuation.get("challenge") or {} if not ch: try: async for ev in run.store.replay(run_id): @@ -3035,6 +3137,15 @@ def _ensure_standby(self, run_id: str, cmd: dict[str, Any]) -> bool: return False if self._standby_busy(run): return False # a standby is already serving this run — don't pile on + action = str(cmd.get("action") or "").lower() + if action in {"ask", "writeup"}: + followup_id = str( + cmd.get("followup_id") + or cmd.get("command_id") + or uuid.uuid4().hex + ) + cmd["followup_id"] = followup_id + run.active_followups.add(followup_id) # A prior driver clears these only after the runtime-exit fence. Clear stale # registrations defensively before publishing the next worker instance. run.standby_cancel = None @@ -3050,6 +3161,22 @@ def _ensure_standby(self, run_id: str, cmd: dict[str, Any]) -> bool: driver = build_standby_driver(cmd, mgr=self) async def _go() -> None: + async def _emit_followup_failed(detail: str) -> None: + if action not in {"ask", "writeup"}: + return + try: + await run.bus.emit(Event( + event_type=EventType.FOLLOWUP_FAILED, + run_id=run_id, + payload={ + "followup_id": cmd.get("followup_id"), + "kind": action, + "detail": detail, + }, + )) + except Exception: + pass + try: LOG.info("standby worker starting for %s action=%s", run_id, cmd.get("action")) @@ -3057,6 +3184,7 @@ async def _go() -> None: LOG.info("standby worker finished for %s action=%s", run_id, cmd.get("action")) except asyncio.CancelledError: + await _emit_followup_failed("后续操作已取消") raise except Exception as exc: detail = _safe_exception_detail("standby worker failed", exc) @@ -3065,17 +3193,20 @@ async def _go() -> None: LOG.error("standby worker failed for %s action=%s error_type=%s", run_id, cmd.get("action"), type(exc).__name__) try: - await run.bus.emit(Event( - event_type=EventType.HITL_REQUEST, - run_id=run_id, - payload={ - "target": cmd.get("target") or "global", - "source": "standby", - "action": cmd.get("action"), - "need": detail, - "text": detail, - }, - )) + if action in {"ask", "writeup"}: + await _emit_followup_failed(detail) + else: + await run.bus.emit(Event( + event_type=EventType.HITL_REQUEST, + run_id=run_id, + payload={ + "target": cmd.get("target") or "global", + "source": "standby", + "action": cmd.get("action"), + "need": detail, + "text": detail, + }, + )) except Exception: pass finally: @@ -3114,6 +3245,9 @@ async def _go() -> None: ] self._ensure_standby_context_cleanup( run, owner=owner, reservations=reservations) + followup_id = str(cmd.get("followup_id") or "") + if followup_id: + run.active_followups.discard(followup_id) run.standby_task = asyncio.create_task(_go()) return True diff --git a/apps/web/server.py b/apps/web/server.py index 6ba21a4..514b37b 100644 --- a/apps/web/server.py +++ b/apps/web/server.py @@ -167,6 +167,7 @@ async def health() -> Any: def llm_settings_payload(config: dict[str, Any]) -> dict[str, Any]: """Expose credential presence/source without returning any secret value.""" from apps.web.llm_credentials import LlmCredentialStore + from muteki.solver.worker_profiles import resolve_seat_ref payload = copy.deepcopy(config) store = LlmCredentialStore(app.state.manager.sessions_root) @@ -175,6 +176,24 @@ def llm_settings_payload(config: dict[str, Any]) -> dict[str, Any]: row = profiles.get(which) if isinstance(row, dict): row["credential_source"] = store.source(which) + # The settings UI edits canonical Seat IDs. Scheduler policy may still be + # stored with a legacy profile name, so translate only the API payload and + # leave the scheduler's legacy projection unchanged. + seats = [row for row in (payload.get("seats") or []) if isinstance(row, dict)] + aliases = payload.get("seat_alias") if isinstance(payload.get("seat_alias"), dict) else {} + coordinator = (payload.get("stage_policy") or {}).get("coordinator") or {} + for key, role in (("review", "review"), ("verifier", "verifier")): + policy = coordinator.get(key) + if not isinstance(policy, dict): + continue + canonical = resolve_seat_ref( + policy.get("engine"), seats=seats, alias_table=aliases) + if canonical is None: + canonical = next(( + str(seat.get("id")) for seat in seats + if role in (seat.get("roles") or []) and seat.get("enabled", True) + ), None) + policy["engine"] = canonical or "" return payload # Auth (P3): a single-password gate in front of /api. fail_fast_check refuses @@ -477,7 +496,6 @@ async def btw(run_id: str, request: Request) -> Any: graph_db = root / "graph" / "shared_graph.db" jsonl_path = (mgr.sessions_root / f"{safe}.jsonl").resolve() board_path = root / ".muteki_board.md" - winner_path = root / "winner.json" arts_path = root / "arts" uploads_path = (mgr.sessions_root / safe / "uploads").resolve() challenge_name = run.name or run_id @@ -496,14 +514,7 @@ async def btw(run_id: str, request: Request) -> Any: limiter = BtwLimiter() app.state.btw_limiters = limiter # type: ignore[attr-defined] - winner: dict[str, Any] = {} - if winner_path.exists(): - try: - raw = json.loads(winner_path.read_text(encoding="utf-8")) - if isinstance(raw, dict): - winner = raw - except Exception: - winner = {} + winner = mgr.load_winner_continuation(run_id) wc = mgr.worker_config.resolve(challenge_category) worker_profiles = wc.get("worker_profiles") or [] worker_network = str(wc.get("worker_network") or "bridge") @@ -599,7 +610,9 @@ def _worker_path(p: Path) -> str: jsonl=_worker_path(jsonl_path), graph_db=_worker_path(graph_db), board=_worker_path(board_path), - winner=_worker_path(winner_path), + # The Worker-writable winner artifact is intentionally not + # supplied as evidence to a side-query worker. + winner="", arts=_worker_path(arts_path), uploads=_worker_path(uploads_path), ), @@ -785,23 +798,31 @@ async def put_worker_settings(request: Request) -> Any: row.pop("clear_api_key", None) row.pop("credential_source", None) try: - cfg = app.state.manager.worker_config.set( - engines=body.get("engines"), - start_workers=body.get("start_workers"), - max_workers=body.get("max_workers"), - worker_backend=body.get("worker_backend"), - worker_network=body.get("worker_network"), - race_scout=body.get("race_scout"), - race_timeout=body.get("race_timeout"), - wall_clock_budget=body.get("wall_clock_budget"), - race_engines=body.get("race_engines"), - max_total_workers=body.get("max_total_workers"), - cost_budget_usd=body.get("cost_budget_usd"), - stage_policy=body.get("stage_policy"), - llm_profiles=llm_profiles, - worker_profiles=body.get("worker_profiles"), - overrides=body.get("overrides"), - ) + settings = { + "engines": body.get("engines"), + "start_workers": body.get("start_workers"), + "max_workers": body.get("max_workers"), + "worker_backend": body.get("worker_backend"), + "worker_network": body.get("worker_network"), + "race_scout": body.get("race_scout"), + "race_timeout": body.get("race_timeout"), + "wall_clock_budget": body.get("wall_clock_budget"), + "race_engines": body.get("race_engines"), + "max_total_workers": body.get("max_total_workers"), + "cost_budget_usd": body.get("cost_budget_usd"), + "stage_policy": body.get("stage_policy"), + "llm_profiles": llm_profiles, + "worker_profiles": body.get("worker_profiles"), + "overrides": body.get("overrides"), + } + if "seats" in body or "credentials" in body: + cfg = app.state.manager.worker_config.set_configuration( + seats=body.get("seats"), + credentials=body.get("credentials"), + **settings, + ) + else: + cfg = app.state.manager.worker_config.set(**settings) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) if isinstance(raw_llm_profiles, dict): diff --git a/apps/web/ui/components/Conversation.tsx b/apps/web/ui/components/Conversation.tsx index 17fe8a2..7a979aa 100644 --- a/apps/web/ui/components/Conversation.tsx +++ b/apps/web/ui/components/Conversation.tsx @@ -71,13 +71,6 @@ const QUICK_RUNNING: Array<{ key: string; labelKey: string; tipKey: string; icon { key: "freeze", labelKey: "quick.freeze", tipKey: "quick.freeze.tip", icon: "lock" }, { key: "thaw", labelKey: "quick.thaw", tipKey: "quick.thaw.tip", icon: "play" }, ]; -// FINISHED — relaunch / converse / wrap up. -const QUICK_FINISHED: Array<{ key: string; labelKey: string; tipKey: string; icon: IconName; primary?: boolean }> = [ - { key: "resolve", labelKey: "quick.resolve", tipKey: "quick.resolve.tip", icon: "play", primary: true }, - { key: "ask", labelKey: "quick.ask", tipKey: "quick.ask.tip", icon: "help" }, - { key: "writeup", labelKey: "quick.writeup", tipKey: "quick.writeup.tip", icon: "pencil" }, -]; - // max height the dispatch textarea auto-grows to (~6–7 rows) before it scrolls // internally; mirrored by `.composer2 textarea { max-height }` in globals.css. const DISPATCH_MAX_H = 180; @@ -181,8 +174,10 @@ function CoordBubble({ }) { const [copied, copy] = useCopied(); const text = m.i18nKey ? t(m.i18nKey, m.i18nVars) : m.content; + const standbyReply = m.role === "agent" && m.solverId?.endsWith("-standby"); const who = m.role === "human" ? t("coord.you") : m.role === "system" - ? (m.kind === "insight" ? "insight" : "system") : t("coord.title"); + ? (m.kind === "insight" ? "insight" : "system") + : standbyReply ? t("coord.solveWorker") : t("coord.title"); const cls = m.role === "human" ? "you" : m.role === "system" ? `system ${m.kind}` : `coordinator ${m.kind}`; // long coordinator reasoning folds to keep the thread scannable const isLong = m.role === "agent" && m.kind === "reasoning" && text.length > 520; @@ -785,6 +780,8 @@ function Composer({ started, solved, running, + finished, + followupPending, paused, solvers, flags, @@ -800,6 +797,8 @@ function Composer({ started: boolean; solved: boolean; running: boolean; + finished: boolean; + followupPending: boolean; paused: boolean; solvers: string[]; flags: string[]; @@ -1262,6 +1261,59 @@ function Composer({ ); } + if (finished) { + const ask = () => { + const question = text.trim(); + if (!question || followupPending) return; + void onCommand("global", "ask", question).then((ok) => { + if (ok) setText(""); + }); + }; + return ( +
+
+
+ setText(e.target.value)} + onKeyDown={(e) => { + if (e.key === "Enter" && !e.shiftKey) { e.preventDefault(); ask(); } + }} + placeholder="输入要向解题 Worker 追问的问题" + /> + +
+
+
+ + + + {solved ? ( + <> + + + + ) : null} +
+
{followupPending ? "正在处理后续操作…" : t("composer.finishedHint")}
+
+ {solved && markFalseOpen && flags.length > 1 ? ( +
{flags.map((flag) => ( + + ))}
+ ) : null} +
+
+ ); + } + return (
@@ -1295,38 +1347,7 @@ function Composer({ ) : ( - <> - {QUICK_FINISHED.map((a) => ( - - ))} - {solved && ( - <> - - - - )} - + 任务正在结束，请等待完成事件 )}
{running ? t("composer.steerHint") : t("composer.finishedHint")}
@@ -1717,6 +1738,8 @@ export function Conversation({ started={deck.started} solved={deck.solved} running={running} + finished={deck.finished} + followupPending={deck.followupPending} paused={digest.phase === "paused"} solvers={solvers} flags={deck.flags} diff --git a/apps/web/ui/components/WorkerOrchestration.tsx b/apps/web/ui/components/WorkerOrchestration.tsx index 396199d..57aded5 100644 --- a/apps/web/ui/components/WorkerOrchestration.tsx +++ b/apps/web/ui/components/WorkerOrchestration.tsx @@ -22,11 +22,11 @@ import { getWorkerModelOptions, getWorkerImageStatus, getWorkerSettings, + getSystemLogin, importHostCodexAuth, importHostWorkerLogin, listCredentialAccounts, putCredentialAccount, - putWorkerIdentity, putWorkerSettings, pullWorkerImage, testCredentialAccount, @@ -181,6 +181,10 @@ function isOrdinarySeat(seat: Seat): boolean { return seat.roles.some((role) => ORDINARY_ROLES.includes(role)); } +function canServeChannel(seat: Seat, role: "review" | "verifier"): boolean { + return isOrdinarySeat(seat) || seat.roles.includes(role); +} + function accountWorkerEngine(account?: CredentialAccount | null): Engine | null { if (!account) return null; const target = account.worker_engine || account.details?.target_engine || account.engine; @@ -1407,7 +1411,7 @@ function ReviewInspector({ testing: boolean; testResult: WorkerModelTestResult | null; }) { - const options = seats.filter((seat) => seat.roles.includes("review")); + const options = seats.filter((seat) => canServeChannel(seat, "review")); const selected = options.find((seat) => seat.id === review.engine); const credential = credentials.find((item) => item.id === selected?.credential_id); const dedicated = Boolean(selected && !isOrdinarySeat(selected)); @@ -1427,7 +1431,7 @@ function ReviewInspector({
启用 Review按触发条件启动独立审查进程 onReview({ enabled })} />
普通并发与 Review 并发相互独立
onReview({ reasoning_effort })} /> @@ -1521,7 +1525,7 @@ function VerifierInspector({ testing: boolean; testResult: WorkerModelTestResult | null; }) { - const options = seats.filter((seat) => seat.roles.includes("verifier")); + const options = seats.filter((seat) => canServeChannel(seat, "verifier")); const selected = options.find((seat) => seat.id === verifier.engine); const credential = credentials.find((item) => item.id === selected?.credential_id); const dedicated = Boolean(selected && !isOrdinarySeat(selected)); @@ -1544,7 +1548,7 @@ function VerifierInspector({
启用 Verifierrace-scout 期间独立复现已提交报告 onVerifier({ enabled })} />
普通并发与 Verifier 并发相互独立
onVerifier({ reasoning_effort })} /> @@ -2023,6 +2027,7 @@ export function WorkerOrchestration() { const [batchCheck, setBatchCheck] = useState({ running: false, completed: 0, total: 0 }); const [discoveringModels, setDiscoveringModels] = useState(false); const [testResults, setTestResults] = useState>({}); + const [systemLogins, setSystemLogins] = useState>({}); const [feedback, setFeedback] = useState(""); const returnTo = useMemo(() => { @@ -2075,7 +2080,7 @@ export function WorkerOrchestration() { useEffect(() => { let alive = true; - Promise.all([getWorkerSettings(), listCredentialAccounts(), getWorkerModelOptions(), fetchProfilesHealth(), getWorkerImageStatus()]).then(([cfg, accountRows, modelRows, healthRows, workerImage]) => { + Promise.all([getWorkerSettings(), listCredentialAccounts(), getWorkerModelOptions(), fetchProfilesHealth(), getWorkerImageStatus(), getSystemLogin()]).then(([cfg, accountRows, modelRows, healthRows, workerImage, loginRows]) => { if (!alive || !cfg) return; const identity = legacyIdentity(cfg); const ordinary = identity.seats.filter(isOrdinarySeat); @@ -2091,6 +2096,7 @@ export function WorkerOrchestration() { setBackend(cfg.worker_backend || "local"); setNetwork(cfg.worker_network || "bridge"); setImageStatus(workerImage); + setSystemLogins(loginRows); setRaceScout(cfg.race_scout); setRaceTimeout(cfg.race_timeout); setStartWorkers(cfg.start_workers); @@ -2169,10 +2175,10 @@ export function WorkerOrchestration() { setSeats(next); setSelectedId(next.find(isOrdinarySeat)?.id || next[0]?.id || null); if (review.engine === id) { - setReview((current) => ({ ...current, engine: next.find((seat) => seat.roles.includes("review") && seat.enabled)?.id || "" })); + setReview((current) => ({ ...current, engine: next.find((seat) => canServeChannel(seat, "review") && seat.enabled)?.id || "" })); } if (verifier.engine === id) { - setVerifier((current) => ({ ...current, engine: next.find((seat) => seat.roles.includes("verifier") && seat.enabled)?.id || "" })); + setVerifier((current) => ({ ...current, engine: next.find((seat) => canServeChannel(seat, "verifier") && seat.enabled)?.id || "" })); } markDirty(); }, [markDirty, review.engine, verifier.engine, seats]); @@ -2260,7 +2266,7 @@ export function WorkerOrchestration() { setTestResults((current) => { const next = { ...current }; delete next[seat.id]; return next; }); const accountId = credential?.kind === "system_inherit" ? "__system__" : credential?.secret_ref || ""; try { - const result: WorkerModelTestResult = connection === "custom_endpoint" && !seat.model?.trim() + const probed: WorkerModelTestResult = connection === "custom_endpoint" && !seat.model?.trim() ? { ok: false, detail: "自定义 API 缺少模型 ID，无法发起真实模型请求", @@ -2280,6 +2286,16 @@ export function WorkerOrchestration() { model: seat.model || "", reasoningEffort: seat.reasoning_effort || "default", }), seat.model || "", backend); + const systemLoginPresent = connection === "system" + && backend === "local" + && systemLogins[engineOf(seat.engine)] === "present"; + const result: WorkerModelTestResult = !probed.ok && systemLoginPresent + ? { + ...probed, + detail: `已检测到系统登录；真实模型请求失败：${probed.detail}`, + layer: probed.layer === "auth" ? "model" : probed.layer, + } + : probed; setTestResults((current) => ({ ...current, [seat.id]: result })); setHealth((current) => ({ ...current, @@ -2300,7 +2316,7 @@ export function WorkerOrchestration() { } finally { setTestingIds((current) => { const next = new Set(current); next.delete(seat.id); return next; }); } - }, [accounts, backend, credentials, showFeedback]); + }, [accounts, backend, credentials, showFeedback, systemLogins]); const testAllSeats = useCallback(async () => { const targets = seats.filter((seat) => seat.enabled && ( @@ -2402,8 +2418,8 @@ export function WorkerOrchestration() { if (!config) return; const enabledOrdinary = seats.filter((seat) => isOrdinarySeat(seat) && seat.enabled); if (!enabledOrdinary.length) { setSaveState("error"); showFeedback("至少需要一个启用的普通 Worker"); return; } - if (review.enabled && !seats.some((seat) => seat.id === review.engine && seat.enabled && seat.roles.includes("review"))) { setSaveState("error"); showFeedback("Review 已启用，但没有指定可用 Worker"); return; } - if (verifier.enabled && !seats.some((seat) => seat.id === verifier.engine && seat.enabled && seat.roles.includes("verifier"))) { setSaveState("error"); showFeedback("Verifier 已启用，但没有指定可用 Worker"); return; } + if (review.enabled && !seats.some((seat) => seat.id === review.engine && seat.enabled && canServeChannel(seat, "review"))) { setSaveState("error"); showFeedback("Review 已启用，但没有指定可用 Worker"); return; } + if (verifier.enabled && !seats.some((seat) => seat.id === verifier.engine && seat.enabled && canServeChannel(seat, "verifier"))) { setSaveState("error"); showFeedback("Verifier 已启用，但没有指定可用 Worker"); return; } const invalidLlm = (["planner", "titler"] as const).find((which) => { const profile = llmProfiles[which]; return !profile.model.trim() || ((profile.connection || (profile.base_url ? "custom_endpoint" : "default")) === "custom_endpoint" && !profile.base_url?.trim()); @@ -2418,11 +2434,11 @@ export function WorkerOrchestration() { if (invalidTemperature) { setSaveState("error"); showFeedback(`${invalidTemperature === "planner" ? "Reason / Planner" : "Titler"} 的 Temperature 需在 0 到 2 之间`); return; } setSaveState("saving"); const normalizedSeats = seats.map((seat, index) => ({ ...seat, priority: (index + 1) * 10, roles: [...seat.roles], capacity: { ...seat.capacity } })); - const identity = await putWorkerIdentity({ seats: normalizedSeats, credentials }); - if (!identity) { setSaveState("error"); showFeedback(backend === "container" ? "保存失败：请检查容器 Worker 是否都绑定了可注入凭据" : "身份配置保存失败"); return; } const refs = enabledOrdinary.map((seat) => seat.id); const raceRefs = enabledOrdinary.filter((seat) => seat.race).map((seat) => seat.id); const saved = await putWorkerSettings({ + seats: normalizedSeats, + credentials, engines: refs, race_engines: raceRefs, start_workers: Math.min(Math.max(1, startWorkers), Math.max(1, maxWorkers)), @@ -2450,7 +2466,7 @@ export function WorkerOrchestration() { budgets: { max_total_workers: maxTotal, cost_budget_usd: costBudget }, }, }); - if (!saved) { setSaveState("error"); showFeedback("阵容策略保存失败"); return; } + if (!saved) { setSaveState("error"); showFeedback(backend === "container" ? "保存失败：请检查容器 Worker 是否都绑定了可注入凭据" : "Worker 配置保存失败"); return; } setConfig(saved); setSeats(normalizedSeats); setLlmProfiles(saved.llm_profiles); diff --git a/apps/web/ui/lib/events.ts b/apps/web/ui/lib/events.ts index 633c9cc..86f42dc 100644 --- a/apps/web/ui/lib/events.ts +++ b/apps/web/ui/lib/events.ts @@ -11,6 +11,9 @@ export enum EventType { RUN_TITLED = "run.titled", RUN_FINISHED = "run.finished", RUN_REOPENED = "run.reopened", + FOLLOWUP_STARTED = "followup.started", + FOLLOWUP_COMPLETED = "followup.completed", + FOLLOWUP_FAILED = "followup.failed", FLAG_ACCEPTED = "flag.accepted", PROJECTION_INCOMPLETE = "projection.incomplete", WORKER_STATUS = "worker.status", @@ -163,6 +166,9 @@ export interface ChatMessage { id: string; role: ChatRole; solverId?: string; + /** Correlates a post-solve pending row with its terminal event. */ + followupId?: string; + followupKind?: string; // True for worker-produced conversational follow-ups (post-solve standby ask / // writeup) that should still appear in the main coordinator thread. The solverId // is kept for activity/diagnostics, but the conversation spine treats it as an @@ -687,6 +693,8 @@ export interface DeckState { started: boolean; preparing: boolean; finished: boolean; + /** Ask/Writeup lifecycle is independent from the completed run lifecycle. */ + followupPending: boolean; // wall-clock bookends (event ts, seconds or ms — normalised at read time). // startedAt = first RUN_STARTED; finishedAt = RUN_FINISHED. A running run has // startedAt but no finishedAt → elapsed is measured against "now". @@ -798,6 +806,7 @@ export function emptyDeck(runId: string): DeckState { started: false, preparing: false, finished: false, + followupPending: false, solved: false, flags: [], invalidatedFlags: [], @@ -2704,6 +2713,79 @@ export function reduce(prev: DeckState, ev: MutekiEvent): DeckState { } break; } + case EventType.FOLLOWUP_STARTED: { + const kind = String(p.kind || "ask"); + const followupId = String(p.followup_id || ""); + if (followupId && s.chat.some((message) => + message.role === "system" && message.followupId === followupId)) { + break; + } + s.followupPending = true; + const question = String(p.question || "").trim(); + if (kind === "ask" && question) { + pushChat(s, { + role: "human", kind: "text", content: question, ts: ev.ts, + }); + } + pushChat(s, { + role: "system", kind: "status", + content: kind === "writeup" ? "正在生成报告…" : "正在回答追问…", + ts: ev.ts, followupId, followupKind: kind, + }); + break; + } + case EventType.FOLLOWUP_COMPLETED: { + s.followupPending = false; + const kind = String(p.kind || "ask"); + const followupId = String(p.followup_id || ""); + const pendingStatus = kind === "writeup" ? "正在生成报告…" : "正在回答追问…"; + const completedStatus = kind === "writeup" ? "报告已生成" : "追问已回答"; + const pendingIndex = [...s.chat].reverse().findIndex( + (message) => message.role === "system" + && (followupId + ? message.followupId === followupId + : message.content === pendingStatus), + ); + if (pendingIndex >= 0) { + const index = s.chat.length - pendingIndex - 1; + s.chat = s.chat.map((message, messageIndex) => messageIndex === index + ? { ...message, content: completedStatus, ts: ev.ts } + : message); + } + const text = String(p.text || "").trim(); + if (text) { + pushChat(s, { + role: "agent", solverId: sid || "standby", mainThread: true, + kind: "text", content: text, ts: ev.ts, sealed: true, + }); + } + break; + } + case EventType.FOLLOWUP_FAILED: { + s.followupPending = false; + const followupId = String(p.followup_id || ""); + const kind = String(p.kind || "ask"); + const pendingStatus = kind === "writeup" ? "正在生成报告…" : "正在回答追问…"; + const failedStatus = `后续操作失败：${String(p.detail || "原因未查明")}`; + const pendingIndex = [...s.chat].reverse().findIndex( + (message) => message.role === "system" + && (followupId + ? message.followupId === followupId + : message.content === pendingStatus), + ); + if (pendingIndex >= 0) { + const index = s.chat.length - pendingIndex - 1; + s.chat = s.chat.map((message, messageIndex) => messageIndex === index + ? { ...message, content: failedStatus, ts: ev.ts } + : message); + } else { + pushChat(s, { + role: "system", kind: "status", content: failedStatus, ts: ev.ts, + followupId, followupKind: kind, + }); + } + break; + } case EventType.RUN_REOPENED: { // The same lifecycle event reopens a run for either "continue solving" or a // false-positive flag invalidation. Keep the operator copy precise. diff --git a/apps/web/ui/lib/i18n.tsx b/apps/web/ui/lib/i18n.tsx index 42a4b46..af6d984 100644 --- a/apps/web/ui/lib/i18n.tsx +++ b/apps/web/ui/lib/i18n.tsx @@ -82,7 +82,7 @@ const STRINGS: Dict = { "toast.dispatchFailed": { zh: "派发失败，请检查后端状态", en: "Dispatch failed — check backend status" }, "toast.workerSpawned": { zh: "已派发 worker", en: "Worker dispatched" }, "toast.workerKilled": { zh: "已停止 worker", en: "Worker stopped" }, - "toast.writeupRequested": { zh: "正在生成复盘…协调器整理后会发到对话里", en: "Generating writeup… the coordinator will post it to the thread" }, + "toast.writeupRequested": { zh: "正在生成复盘…获胜 Worker 完成后会发到对话里", en: "Generating writeup… the winning worker will post it to the thread" }, "toast.renamed": { zh: "已重命名", en: "Renamed" }, "toast.deleted": { zh: "已删除", en: "Deleted" }, "toast.folderCreated": { zh: "已新建文件夹", en: "Folder created" }, @@ -425,12 +425,12 @@ const STRINGS: Dict = { "control.status.unknown": { zh: "效果未知", en: "unknown" }, "control.status.rejected": { zh: "已拒绝", en: "rejected" }, "quick.resolve.tip": { zh: "拉起完整蜂群续解", en: "Relaunch the full swarm to keep solving" }, - "quick.ask.tip": { zh: "向协调器提问", en: "Ask the coordinator a question" }, + "quick.ask.tip": { zh: "向获胜 Worker 追问", en: "Ask the winning worker a question" }, "quick.writeup.tip": { zh: "生成复盘报告", en: "Generate a writeup report" }, "quick.markFalse": { zh: "标记误报", en: "Mark false" }, "quick.markFalseTitle": { zh: "这个 flag 是假的 → 重开解题（单 worker 续解）", en: "This flag is wrong → reopen (single-worker re-solve)" }, "quick.writeup": { zh: "生成复盘", en: "Writeup" }, - "quick.ask": { zh: "对话", en: "Ask" }, + "quick.ask": { zh: "追问", en: "Ask" }, "quick.resolve": { zh: "继续做题", en: "Keep solving" }, "quick.resolveTitle": { zh: "重新拉起完整蜂群继续解题（沿用已验证的事实，多 worker 并行）", en: "Relaunch the full swarm to keep solving (reuses verified facts, multi-worker)" }, "quick.stop": { zh: "停止", en: "Stop" }, @@ -767,6 +767,7 @@ const STRINGS: Dict = { // ---- coordinator thread (main conversation) ---- "coord.title": { zh: "协调器", en: "Coordinator" }, + "coord.solveWorker": { zh: "解题 Worker", en: "Solve worker" }, "coord.subtitle": { zh: "DeepSeek 规划 · 审计 · 判决", en: "DeepSeek plans · audits · adjudicates" }, "coord.you": { zh: "你", en: "you" }, "coord.empty": { zh: "派发题目后，协调器会在这里规划与汇报进展。", en: "After you dispatch, the coordinator plans and reports progress here." }, diff --git a/apps/web/worker_config.py b/apps/web/worker_config.py index 16ed7ff..c0bf55e 100644 --- a/apps/web/worker_config.py +++ b/apps/web/worker_config.py @@ -14,6 +14,7 @@ from __future__ import annotations +import copy import json from pathlib import Path from typing import Any, Optional @@ -63,7 +64,9 @@ "max_review_workers": 12, } DEFAULT_VERIFIER_POLICY = { - "enabled": True, + # Verification requires an explicitly assigned verifier seat. Keeping this + # disabled avoids a default policy that is enabled with an empty foreign key. + "enabled": False, "engine": "", "reasoning_effort": "inherit", "max_concurrent": 0, @@ -830,10 +833,17 @@ def set_identity_model( if not isinstance(credentials, list): raise ValueError("credentials must be a list") self._data["credentials"] = [c for c in credentials if isinstance(c, dict)] - # A container Worker may not use a system_inherit credential because the - # host login is not mounted into the container. + self._validate_identity_backend( + self._clean_backend(self._data.get("worker_backend"))) + # keep the legacy projection in sync so get()/scheduler see the change. + self._project_identity_to_legacy() + self._sync_worker_counts(link_profile_capacity=True) + self._flush() + return self.get() + + def _validate_identity_backend(self, backend: str) -> None: + """Validate enabled seats against one final target backend.""" cred_by_id = {str(c.get("id")): c for c in self._data.get("credentials") or []} - backend = self._clean_backend(self._data.get("worker_backend")) for s in self._data.get("seats") or []: # Disabled seats are retained and never dispatched. # They may keep their host-login binding while an active container @@ -848,11 +858,41 @@ def set_identity_model( f"非法组合:Agent「{label}」在容器环境下使用了「系统登录」凭据。" f"容器不挂载宿主登录态,请改用引擎凭据或自定义端点。" ) - # keep the legacy projection in sync so get()/scheduler see the change. - self._project_identity_to_legacy() - self._sync_worker_counts(link_profile_capacity=True) - self._flush() - return self.get() + + def set_configuration( + self, + *, + seats: Any, + credentials: Any, + **settings: Any, + ) -> dict[str, Any]: + """Validate and persist identity, runtime and policy as one final state. + + ``set()`` performs the only file replacement. Any validation failure + restores the in-memory snapshot, so callers never observe a half-saved + backend/identity combination. + """ + if not isinstance(seats, list): + raise ValueError("seats must be a list") + if not isinstance(credentials, list): + raise ValueError("credentials must be a list") + previous = copy.deepcopy(self._data) + try: + self._data["seats"] = [copy.deepcopy(s) for s in seats if isinstance(s, dict)] + self._data["credentials"] = [ + copy.deepcopy(c) for c in credentials if isinstance(c, dict) + ] + self._project_identity_to_legacy() + target_backend = self._require_backend( + settings.get("worker_backend") + if settings.get("worker_backend") is not None + else self._clean_backend(self._data.get("worker_backend")) + ) + self._validate_identity_backend(target_backend) + return self.set(**settings) + except Exception: + self._data = previous + raise def _sync_worker_counts(self, *, link_profile_capacity: bool) -> None: # Direction is roster→max (the operator owns per-seat capacity; the global diff --git a/muteki/core/events.py b/muteki/core/events.py index b1d992b..90987a7 100644 --- a/muteki/core/events.py +++ b/muteki/core/events.py @@ -25,6 +25,9 @@ class EventType(str, Enum): RUN_REOPENED = "run.reopened" # a terminal run was re-opened (continue solving # or flag marked false-positive) — rail flips solved/finished→running FLAG_ACCEPTED = "flag.accepted" # CAS/outbox-verified Protocol 2 publication; + FOLLOWUP_STARTED = "followup.started" # post-run Ask/Writeup lifecycle; these + FOLLOWUP_COMPLETED = "followup.completed" # events never reopen/finish the run + FOLLOWUP_FAILED = "followup.failed" # accepted-only visibility does not imply solved, finished, or clean closure PROJECTION_INCOMPLETE = "projection.incomplete" # non-terminal, redacted startup # reconciliation diagnostic; never carries candidate/CAS/credential material diff --git a/muteki/solver/btw.py b/muteki/solver/btw.py index 722b6f9..e352d5e 100644 --- a/muteki/solver/btw.py +++ b/muteki/solver/btw.py @@ -152,9 +152,10 @@ def build_btw_worker_prompt( f"- event log JSONL: {paths.jsonl}", f"- shared graph SQLite DB: {paths.graph_db}", f"- markdown board: {paths.board}", - f"- winner snapshot: {paths.winner}", f"- artifacts dir: {paths.arts}", ] + if paths.winner: + path_lines.append(f"- winner snapshot: {paths.winner}") if paths.uploads: path_lines.append(f"- uploaded challenge files: {paths.uploads}") diff --git a/muteki/solver/cli_driver.py b/muteki/solver/cli_driver.py index 2aa6db6..30172ac 100644 --- a/muteki/solver/cli_driver.py +++ b/muteki/solver/cli_driver.py @@ -244,6 +244,10 @@ class CliResult: output_tokens: Optional[int] = None num_turns: Optional[int] = None elapsed_s: float = 0.0 + # Real subprocess exit status. Parsers normalize vendor output and cannot infer + # process success from response text alone, so execution layers attach this + # after the child exits. None means the runner could not observe an exit code. + returncode: Optional[int] = None timed_out: bool = False # OOM-killed: the worker's process was SIGKILL'd by the kernel out-of-memory # killer (a sibling run's container ballooned and starved the Docker VM — no @@ -596,7 +600,7 @@ def health_detail(self, *, env: "dict[str, str] | None" = None) -> "tuple[bool, class ClaudeCodeDriver(CliDriver): - """`claude -p` — pre-seeds a uuid session; resumes with `-r`. Bare host, + """`claude -p` — pre-seeds a uuid session; resumes with `-r`. Host CLI, --dangerously-skip-permissions (full shell), JSON output for clean parsing.""" name = "claude" secure_prompt_transport = True @@ -658,14 +662,14 @@ def build_execute_stdin( ) -> list[str]: # Claude print mode reads text input from stdin when no positional prompt is # supplied. --no-session-persistence is the vendor-supported disk fence; - # --bare also disables auto-memory, hooks/plugin sync, background prefetches, - # and keychain reads that are unnecessary for an env-authenticated worker. + # ProfileDriver adds --bare for injected credentials and endpoints. The + # base driver represents host system login and must retain Keychain access. # Keep a trailing `--` sentinel so profile model injection has an unambiguous # insertion point, but never put the prompt itself in argv. del prompt, session return [ self.bin, "-p", *self._fmt(stream), - "--dangerously-skip-permissions", "--bare", + "--dangerously-skip-permissions", "--no-session-persistence", *self._mcp_isolation(kb_access=kb_access), *self._denied(web_access=web_access, kb_access=kb_access), @@ -675,7 +679,7 @@ def build_execute_stdin( def secure_prompt_preflight(self) -> "tuple[bool, str]": return _secure_help_preflight( self.bin, ["--help"], - ("--no-session-persistence", "--bare", "--print")) + ("--no-session-persistence", "--print")) def build_resume( self, prompt: str, session: str, *, @@ -826,12 +830,11 @@ def parse_stream_steps(self, line: str) -> list[StreamStep]: def _hello_argv(self) -> list[str]: # one-turn JSON dry-run; _hello_ok asserts the result envelope came back. - # Match build_execute_stdin's --bare / --no-session-persistence fence so a - # host plugin SessionEnd hook (or missing `node`) cannot false-fail the - # coordinator health probe and empty the entire engine roster. + # Keep the minimal turn non-persistent. ProfileDriver adds --bare only for + # injected credentials/endpoints; host system login needs Keychain reads. return [ self.bin, "-p", "--output-format", "json", "--max-turns", "1", - "--dangerously-skip-permissions", "--bare", "--no-session-persistence", + "--dangerously-skip-permissions", "--no-session-persistence", *self._mcp_isolation(kb_access=False), "--tools", "", "--", self.HELLO_PROMPT, @@ -2428,7 +2431,15 @@ def _reasoning_effort(self) -> str: self.profile.get("reasoning_effort"), "default") def _with_profile_options(self, argv: list[str]) -> list[str]: - out = _insert_model_arg(argv, self._model(), engine=self.name) + out = list(argv) + credential_kind = str( + self.profile.get("credential_kind") + or ("engine_key" if self.profile.get("credential_account") else "system_inherit") + ).strip() + if self.name == "claude" and credential_kind != "system_inherit" and "--bare" not in out: + sentinel = out.index("--") if "--" in out else len(out) + out.insert(sentinel, "--bare") + out = _insert_model_arg(out, self._model(), engine=self.name) return apply_reasoning_effort( out, engine=self.name, reasoning_effort=self._reasoning_effort()) @@ -2523,7 +2534,10 @@ def env_extra(self) -> "dict[str, str]": return env def _with_profile_options(self, argv: list[str]) -> list[str]: - out = argv + out = list(argv) + if self.name == "claude" and "--bare" not in out: + sentinel = out.index("--") if "--" in out else len(out) + out.insert(sentinel, "--bare") if not (self.name == "codex" and self._codex_config_flags()): selected_model = str(self.profile.get("model") or "").strip() if self.name == "opencode" and selected_model and "/" not in selected_model: @@ -3512,6 +3526,7 @@ def run_cli(driver: CliDriver, argv: list[str], *, cwd: str, timeout: int, res.elapsed_s = time.time() - t0 return res res = driver.parse(proc.stdout or "", proc.stderr or "") + res.returncode = proc.returncode res.elapsed_s = time.time() - t0 return res @@ -3833,6 +3848,7 @@ def _drain_stderr() -> None: if on_raw_streams is not None: on_raw_streams(stdout, stderr) res = driver.parse(stdout, stderr or "") + res.returncode = proc.returncode res.timed_out = timed_out res.cancelled = cancelled res.steered = steered diff --git a/muteki/solver/cli_solver.py b/muteki/solver/cli_solver.py index 63e0da9..fb7e8a7 100644 --- a/muteki/solver/cli_solver.py +++ b/muteki/solver/cli_solver.py @@ -784,6 +784,8 @@ def _looks_like_verifier_output(text: str) -> bool: _RESPOND_WRITEUP_PROMPT = ( "Write a concise CTF WRITEUP for the challenge you just solved, in Chinese. " "Base it ONLY on what you actually confirmed this session — do not invent steps. " + "Do not run commands, call tools, search the filesystem, or continue the " + "investigation. Synthesize the report from the confirmed session history now. " "Structure it as:\n" " ## 漏洞点 (the root cause / vulnerability)\n" " ## 利用步骤 (numbered, reproducible — the real commands/requests you used)\n" @@ -791,6 +793,15 @@ def _looks_like_verifier_output(text: str) -> bool: "Keep it tight and technical. Output ONLY the markdown writeup, nothing else." ) +_RESPOND_PENTEST_WRITEUP_PROMPT = ( + "Write a concise penetration-testing report in Chinese for the engagement you " + "just completed. Base it ONLY on evidence confirmed this session. " + "Do not run commands, call tools, search the filesystem, or continue the " + "investigation. Synthesize the report from the confirmed session history now. " + "Structure it as:\n ## 范围与目标\n ## 已确认发现\n ## 复现步骤\n ## 影响与修复建议\n" + "Do not invent findings, commands, requests, or impact. Output ONLY markdown." +) + class CliSolver: """Swarm worker backed by a shelled CLI agent. Mirrors Solver's interface.""" @@ -6604,11 +6615,20 @@ async def _run_respond(self) -> SolveOutcome: keeps going and any NEW flag it finds STILL passes the real gate.""" action = (self.hitl_cmd.get("action") or "ask").lower() text = (self.hitl_cmd.get("text") or "").strip() - await self._emit(EventType.RUN_STARTED, challenge=self.challenge.model_dump()) - await self._emit( - EventType.REASONING_DELTA, - text=f"[{self.driver.name}] standby — resuming session for " - f"{action}{(': ' + text[:80]) if text else ''}\n") + followup_id = str(self.hitl_cmd.get("followup_id") or "") + if action in {"ask", "writeup"}: + await self._emit( + EventType.FOLLOWUP_STARTED, + followup_id=followup_id, + kind=action, + question=text, + ) + else: + await self._emit(EventType.RUN_STARTED, challenge=self.challenge.model_dump()) + await self._emit( + EventType.REASONING_DELTA, + text=f"[{self.driver.name}] standby — resuming session for " + f"{action}{(': ' + text[:80]) if text else ''}\n") # per-worker cwd: reuse the winner's workdir if it still exists (keeps any # files it downloaded), else a fresh scratch dir. Computed FIRST so we can @@ -6629,7 +6649,11 @@ async def _run_respond(self) -> SolveOutcome: prompt = _RESPOND_MARK_FALSE_PROMPT.format( flag=self.hitl_cmd.get("flag") or "(the reported flag)", note=note) elif action == "writeup": - prompt = _RESPOND_WRITEUP_PROMPT + prompt = ( + _RESPOND_PENTEST_WRITEUP_PROMPT + if str(getattr(self.challenge, "mode", "ctf")) == "pentest" + else _RESPOND_WRITEUP_PROMPT + ) else: # ask / hint / redirect / anything conversational question = text or "(no question text)" if action == "redirect": @@ -6678,16 +6702,41 @@ async def _run_respond(self) -> SolveOutcome: await self._note_cli_session(session) argv, stdin_text = self._execute_invocation(prompt, session) + # Conversational follow-ups should return promptly. A resumed model can + # otherwise start another investigation and leave the deck pending for the + # normal 40-minute solve timeout. mark_false remains a real re-solve and + # retains the longer bound. + respond_timeout = 300 if action in {"ask", "writeup"} else 1200 res: CliResult = await self._run_invocation( - argv, cwd=str(wd), timeout=min(self.timeout, 1200), + argv, cwd=str(wd), timeout=min(self.timeout, respond_timeout), stdin_text=stdin_text) await self._emit_empty_stderr_diagnostic(res) await self._stream_cost(res) all_text = self._result_text_with_stderr(res) safe_all_text = self._redact_control_secrets(all_text) + if action in {"ask", "writeup"}: + observed_rc = res.returncode + if observed_rc is None: + runtime_rc = (res.runtime_status or {}).get("rc") + try: + observed_rc = int(runtime_rc) if runtime_rc is not None else None + except (TypeError, ValueError): + observed_rc = None + if res.timed_out: + raise RuntimeError(f"standby {action} worker timed out") + if res.oom_killed: + raise RuntimeError(f"standby {action} worker was terminated by OOM") + if res.cancelled or res.steered: + raise RuntimeError(f"standby {action} worker did not complete") + if observed_rc not in {None, 0}: + raise RuntimeError( + f"standby {action} worker exited with code {observed_rc}") + if not safe_all_text.strip(): + raise RuntimeError(f"standby {action} worker returned an empty response") + # stream the reply to the deck (the worker's answer / writeup body). - if safe_all_text.strip(): + if safe_all_text.strip() and action not in {"ask", "writeup"}: await self._emit( EventType.TEXT_MESSAGE_DELTA, text=safe_all_text.strip(), diff --git a/muteki/solver/container_exec.py b/muteki/solver/container_exec.py index cfc179e..2e00dcf 100644 --- a/muteki/solver/container_exec.py +++ b/muteki/solver/container_exec.py @@ -1008,9 +1008,12 @@ def run_cli_container(driver: CliDriver, argv: list[str], *, handle: ContainerHa res = run_cli_rcp(driver, cont_argv, run_id=handle.run_id, container_cwd=cont_cwd, timeout=timeout, env=env, **rcp_kwargs) + observed_rc = (res.runtime_status or {}).get("rc") + if res.returncode is None and observed_rc is not None: + res.returncode = int(observed_rc) status = ("oom" if res.oom_killed else "timeout" if res.timed_out else "finished") res.runtime_status = _RUNTIME_REGISTRY.finish( - rec, status=status, rc=(res.runtime_status or {}).get("rc"), + rec, status=status, rc=observed_rc, timed_out=res.timed_out, oom_killed=res.oom_killed, error=(res.raw_stderr or "").strip()[:300]) return res @@ -1077,6 +1080,8 @@ def _on_proc(proc: object) -> None: paused_event=paused_event, **rcp_kwargs) rs = res.runtime_status or {} + if res.returncode is None and rs.get("rc") is not None: + res.returncode = int(rs["rc"]) res.runtime_status = _RUNTIME_REGISTRY.finish( rec, status=rs.get("status", "finished"), rc=rs.get("rc"), timed_out=res.timed_out, oom_killed=res.oom_killed, @@ -1242,6 +1247,7 @@ def run(driver: CliDriver, argv: list[str], *, handle: ContainerHandle, rec, status="timeout", timed_out=True, error="host timeout") return res res = driver.parse(proc.stdout or "", proc.stderr or "") + res.returncode = proc.returncode if proc.returncode == 137: oom_after = _oom_kill_count(handle.container) if (oom_before is not None and oom_after is not None @@ -1435,6 +1441,7 @@ def _watch() -> None: except Exception: pass res = driver.parse("".join(out_lines), stderr or "") + res.returncode = rc res.timed_out = timed_out res.oom_killed = oom_killed res.cancelled = cancelled diff --git a/muteki/solver/identity_model.py b/muteki/solver/identity_model.py index a4d1eeb..d9e3d1e 100644 --- a/muteki/solver/identity_model.py +++ b/muteki/solver/identity_model.py @@ -370,6 +370,9 @@ def seat_to_legacy_profile( "credential_mode": credential_mode, "auth": credential_mode, "credential_account": credential_account, + # Preserve the canonical source so command construction can distinguish + # host system login from injected credentials. + "credential_kind": kind, "api_key_ref": "", "base_url": base_url, "wire_api": wire_api or ("responses" if engine == "codex" and base_url else ""), diff --git a/muteki/solver/types.py b/muteki/solver/types.py index ed685f6..592abc6 100644 --- a/muteki/solver/types.py +++ b/muteki/solver/types.py @@ -7,7 +7,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Optional +from typing import Any, Optional from muteki.models.solve_graph import SolveGraph @@ -55,3 +55,6 @@ class SolveOutcome: # so the standby driver can persist it (e.g. writeup.md) without re-parsing the # event stream. Empty for solve runs. reply: str = "" + # Non-secret profile snapshot used to resume the exact winning seat after a + # settings change or server restart. + runtime_profile: dict[str, Any] = field(default_factory=dict) diff --git a/muteki/swarm/coordinator_flags.py b/muteki/swarm/coordinator_flags.py index c9d94a9..1300331 100644 --- a/muteki/swarm/coordinator_flags.py +++ b/muteki/swarm/coordinator_flags.py @@ -1223,22 +1223,16 @@ def _cleanup_finished_worker_dirs(self) -> None: """Remove failed/finished worker scratch while preserving durable run data. The workspace root keeps shared/, inputs/, graph/, final/, manifest.json, - and winner.json. Only non-winner worker cwd directories under workers/ are + and winner.json. Only non-winner worker cwd directories under workers/ are removed at run finish to avoid long coordinator runs accumulating hundreds of duplicate scratch trees. """ if self.worker_root is None: return - keep: list[str] = [] - if self.workspace_root is not None: - winner = self.workspace_root / "winner.json" - try: - data = json.loads(winner.read_text(encoding="utf-8")) - workdir = data.get("workdir") - if workdir: - keep.append(Path(str(workdir)).name) - except Exception: - pass + winner_workdir_name = str( + getattr(self, "_winner_workdir_name", "") or "" + ).strip() + keep = [winner_workdir_name] if winner_workdir_name else [] cleanup_worker_scratch(self.worker_root, keep=keep) @staticmethod diff --git a/muteki/swarm/coordinator_loop.py b/muteki/swarm/coordinator_loop.py index deafac2..ec3fdc1 100644 --- a/muteki/swarm/coordinator_loop.py +++ b/muteki/swarm/coordinator_loop.py @@ -2766,9 +2766,12 @@ def _persist_winner( self, outcome: "Optional[SolveOutcome]", flag: "Optional[str]", *, worker_id: str = "", ) -> None: - """Write the winner's CLI continuation handle to workspace/winner.json so a - post-solve standby driver can resume the SAME session for a human - follow-up. Best-effort: a write failure must never fail a solved run. + """Persist the winner's CLI continuation handle for human follow-ups. + + The Web driver installs a coordinator-only writer. ``winner.json`` remains + a compatibility artifact in the Worker workspace and carries no profile, + credential endpoint or backend authority. Best-effort: a write failure must + never fail a solved run. Needs graph_dir (web runs) — winner.json lands beside graph/ (a sibling of the sandbox root, so sandbox.shutdown_all()'s rmtree can't delete it). TUI @@ -2781,19 +2784,37 @@ def _persist_winner( return try: import json - payload = { + workdir = getattr(outcome, "workdir", "") or "" + self._winner_workdir_name = Path(workdir).name if workdir else "" + trusted_payload = { "engine": getattr(outcome, "engine", "") or "", "worker_id": str(worker_id or ""), "session": session, - "workdir": getattr(outcome, "workdir", "") or "", + "workdir": workdir, "flag": flag or outcome.flag or "", # multi-flag: every flag the run collected (the run's authoritative # set, not just this one worker's). `flag` stays the first. "flags": list(self._found_flags) or ( [flag] if flag else (outcome.flags or [])), "challenge": self.challenge.model_dump(), + "profile": dict(getattr(outcome, "runtime_profile", {}) or {}), **self._runtime_metadata_for(outcome), } + writer = getattr(self, "_winner_continuation_writer", None) + if callable(writer): + writer(dict(trusted_payload)) + profile = trusted_payload.get("profile") or {} + payload = { + key: trusted_payload[key] + for key in ( + "engine", "worker_id", "session", "workdir", "flag", + "flags", "challenge", + ) + } + if isinstance(profile, dict): + payload["profile_id"] = str( + profile.get("id") or profile.get("name") or "" + ) dest = self._graph_dir.parent / "winner.json" dest.parent.mkdir(parents=True, exist_ok=True) dest.write_text(json.dumps(payload, ensure_ascii=False, indent=2)) diff --git a/muteki/swarm/coordinator_race.py b/muteki/swarm/coordinator_race.py index bb639a9..baf35bd 100644 --- a/muteki/swarm/coordinator_race.py +++ b/muteki/swarm/coordinator_race.py @@ -1078,7 +1078,19 @@ def _take_context_secret_values( async def _run_control_worker(self, worker: Any) -> Any: """Run a worker; CliSolver commits reserved context from ``_on_proc``.""" - return await worker.run() + outcome = await worker.run() + profile = getattr(getattr(worker, "driver", None), "profile", None) + if isinstance(profile, dict): + outcome.runtime_profile = { + key: profile.get(key) + for key in ( + "id", "name", "label", "engine", "transport", "model", + "reasoning_effort", "credential_account", "credential_kind", + "credential_mode", "base_url", "wire_api", + ) + if profile.get(key) not in (None, "") + } + return outcome async def _schedule_control_worker( self, worker: Any, *, name: str, intent_id: str = "", diff --git a/muteki/swarm/swarm.py b/muteki/swarm/swarm.py index 3da70d4..44ebe47 100644 --- a/muteki/swarm/swarm.py +++ b/muteki/swarm/swarm.py @@ -704,7 +704,8 @@ def __init__( else: db_path = self.sandbox.root / self.run_id / "shared_graph.db" # remember where durable per-run state lives (sibling of graph/) so a - # post-solve standby can find winner.json + the shared graph again. + # post-solve standby can find private continuation state and the + # shared graph again. self._graph_dir = Path(graph_dir) if graph_dir is not None else None self.shared_graph = SQLiteSharedGraph.open( db_path=db_path, diff --git a/tests/test_cli_executor.py b/tests/test_cli_executor.py index ec9d615..c1cf51e 100644 --- a/tests/test_cli_executor.py +++ b/tests/test_cli_executor.py @@ -205,7 +205,7 @@ def test_secret_prompt_invocations_are_stdin_only_and_non_persistent(): claude_argv = claude.build_execute_stdin(secret, claude.new_session(), stream=True) assert secret not in "\0".join(claude_argv) assert "--no-session-persistence" in claude_argv - assert "--bare" in claude_argv + assert "--bare" not in claude_argv assert claude_argv[-1] == "--" codex = CodexDriver() @@ -220,6 +220,14 @@ def test_secret_prompt_invocations_are_stdin_only_and_non_persistent(): wrapped_argv = wrapped.build_execute_stdin(secret, None, stream=True) assert secret not in "\0".join(wrapped_argv) assert wrapped_argv[wrapped_argv.index("--model") + 1] == "x" + assert "--bare" not in wrapped_argv + + injected = driver_for({ + "id": "claude-key", "engine": "claude", "model": "x", + "credential_kind": "engine_key", "credential_account": "claude-main", + }) + injected_argv = injected.build_execute_stdin(secret, None, stream=True) + assert "--bare" in injected_argv # Cursor's installed CLI reads a missing headless positional from stdin, but # exposes no no-persistence mode; exact secret delivery must fail closed. @@ -5272,6 +5280,40 @@ async def _finished(_argv, *, cwd, timeout, stdin_text=None): assert "new target endpoint" in captured["prompt"] +def test_standby_writeup_forbids_new_investigation_and_uses_short_timeout(tmp_path): + from muteki.solver.cli_driver import CliResult + + captured = {} + + class _CaptureResumeDriver(_StubDriver): + def build_resume(self, prompt, *args, **kwargs): + captured["prompt"] = prompt + return ["true"] + + ch = Challenge( + id="writeup-standby", name="writeup", category="web", + flag_format=r"flag\{.*?\}", + ) + solver = CliSolver( + None, ch, bus=_CaptureBus(), driver=_CaptureResumeDriver(""), + engine="claude", kb=False, mode="respond", resume_session="sess", + workdir=str(tmp_path), timeout=2400, + hitl_cmd={"action": "writeup", "followup_id": "F-writeup"}, + ) + + async def _finished(_argv, *, cwd, timeout, stdin_text=None): + captured["timeout"] = timeout + return CliResult(text="## 漏洞点\n已确认。", session="sess") + + solver._run_streaming = _finished + outcome = asyncio.run(solver._run_respond()) + + assert "Do not run commands" in captured["prompt"] + assert "search the filesystem" in captured["prompt"] + assert captured["timeout"] == 300 + assert outcome.reply.startswith("## 漏洞点") + + def test_standby_resume_releases_optional_context_absent_from_prompt(tmp_path): """A resume Popen cannot commit a context omitted from that exact turn.""" from muteki.solver.cli_driver import CliResult diff --git a/tests/test_standby_hitl.py b/tests/test_standby_hitl.py index c36e951..eba885b 100644 --- a/tests/test_standby_hitl.py +++ b/tests/test_standby_hitl.py @@ -34,14 +34,18 @@ def test_persist_winner_writes_session_handle(tmp_path): ) out = SolveOutcome(True, "csawctf{x}", 1, None, "solved", session="sess-abc", engine="claude", workdir="/tmp/w") + trusted = {} + sw._winner_continuation_writer = trusted.update sw._persist_winner(out, "csawctf{x}") winner = json.loads((graph_dir.parent / "winner.json").read_text()) assert winner["session"] == "sess-abc" assert winner["engine"] == "claude" - assert winner["backend"] == "local" - assert "runtime_degraded" in winner + assert "backend" not in winner + assert "profile" not in winner assert winner["flag"] == "csawctf{x}" assert winner["challenge"]["id"] == "run-x" + assert trusted["backend"] == "local" + assert "runtime_degraded" in trusted # multi-flag: winner.json also carries the full flags list (here just the one) assert winner["flags"] == ["csawctf{x}"] @@ -82,6 +86,38 @@ def test_persist_winner_skips_without_session(tmp_path): assert not (graph_dir.parent / "winner.json").exists() +def test_private_winner_continuation_keeps_minimal_trusted_state(tmp_path): + from apps.web.run_manager import RunManager + + mgr = RunManager(sessions_root=tmp_path / "sessions") + workdir = mgr.workspace_dir("run-x") / "workers" / "cli-codex-1" + workdir.mkdir(parents=True) + mgr.persist_winner_continuation("run-x", { + "worker_id": "cli-codex-1", + "engine": "codex", + "session": "thread-1", + "workdir": str(workdir), + "backend": "container", + "profile": { + "id": "seat-codex", + "credential_account": "codex-main", + "base_url": "https://private.example/v1", + }, + "flag": "flag{ok}", + "challenge": {"name": "t", "category": "web"}, + }) + + path = mgr._winner_continuation_path("run-x") + continuation = mgr.load_winner_continuation("run-x") + assert not path.is_relative_to(mgr.workspace_dir("run-x")) + assert path.stat().st_mode & 0o777 == 0o600 + assert continuation["profile_id"] == "seat-codex" + assert continuation["workdir_rel"] == "workers/cli-codex-1" + assert "profile" not in continuation + assert "credential_account" not in path.read_text() + assert "private.example" not in path.read_text() + + # ── D2: false-positive state machine ──────────────────────────────────────── def test_reopen_after_false_positive(tmp_path): g = SQLiteSharedGraph(str(tmp_path / "g.db"), _challenge()) @@ -239,6 +275,46 @@ async def _run(): asyncio.run(_run()) +def test_completed_generation_admits_only_registered_followup_events(tmp_path): + from apps.web import run_manager as rm + from muteki.core.events import Event, EventType + + async def _run(): + mgr = rm.RunManager(sessions_root=str(tmp_path / "sessions")) + run = mgr.create("run-x") + await run.bus.emit(Event( + event_type=EventType.RUN_FINISHED, + run_id="run-x", + payload={"solved": True}, + )) + run.active_followups.add("followup-1") + await run.bus.emit(Event( + event_type=EventType.FOLLOWUP_STARTED, + run_id="run-x", + payload={"followup_id": "followup-1", "kind": "ask"}, + )) + await run.bus.emit(Event( + event_type=EventType.TEXT_MESSAGE_DELTA, + run_id="run-x", + payload={"text": "late runtime frame"}, + )) + await run.bus.emit(Event( + event_type=EventType.FOLLOWUP_COMPLETED, + run_id="run-x", + payload={ + "followup_id": "followup-1", "kind": "ask", "text": "answer", + }, + )) + return [event async for event in run.store.replay("run-x")] + + events = asyncio.run(_run()) + assert [event.event_type for event in events] == [ + EventType.RUN_FINISHED, + EventType.FOLLOWUP_STARTED, + EventType.FOLLOWUP_COMPLETED, + ] + + def test_rehydrated_run_bus_continues_after_persisted_stream_seq(tmp_path): from apps.web import run_manager as rm from muteki.core.events import Event, EventType @@ -590,9 +666,11 @@ async def _run(): (home / "session.txt").write_text("thread-1") wp = mgr.workspace_dir("run-x") / "winner.json" wp.write_text(json.dumps({ - "engine": "codex", - "session": "thread-1", - "workdir": str(mgr.workspace_dir("run-x") / "workers" / "cli-codex-1"), + "engine": "claude", + "profile": {"id": "attacker", "base_url": "https://attacker.invalid"}, + "backend": "local", + "session": "attacker-session", + "workdir": "/tmp/attacker-workdir", "flag": "flag{ok}", "flags": ["flag{ok}"], "challenge": { @@ -600,8 +678,25 @@ async def _run(): "name": "t", "category": "web", "description": "", + }, + })) + trusted_workdir = ( + mgr.workspace_dir("run-x") / "workers" / "cli-codex-1" + ) + trusted_workdir.mkdir(parents=True, exist_ok=True) + mgr.persist_winner_continuation("run-x", { + "engine": "codex", + "profile_id": "seat-codex", + "session": "thread-1", + "workdir": str(trusted_workdir), + "backend": "container", + "flag": "flag{ok}", + "flags": ["flag{ok}"], + "challenge": { + "id": "run-x", "name": "t", "category": "web", + "description": "", }, - })) + }) await run.bus.close() ok = await mgr.post_hitl("run-x", "global", "writeup", text="") assert run.standby_task is not None @@ -628,7 +723,7 @@ async def _run(): assert captured["solver_kwargs"]["resume_session"] == "thread-1" assert captured["solver_kwargs"]["worker_env"]["MUTEKI_WORKER_MODEL"] == "gpt-5.4" assert captured["solver_kwargs"]["worker_env"]["HOME"].endswith("/cli-codex") - assert captured["chown"] == ["standby-codex", "cli-codex"] + assert captured["chown"] == ["cli-codex-1", "cli-codex"] assert captured["teardown"] == {"run_id": "run-x", "remove": True} assert captured["cancel_calls"] >= 2 @@ -713,12 +808,67 @@ async def _run(): caplog.set_level("INFO") seen = asyncio.run(_run()) assert any("standby worker failed" in r.message for r in caplog.records) - reqs = [e for e in seen if e.event_type is EventType.HITL_REQUEST] - assert reqs - assert reqs[-1].payload["need"] == ( + failures = [e for e in seen if e.event_type is EventType.FOLLOWUP_FAILED] + assert failures + assert failures[-1].payload["followup_id"] + assert failures[-1].payload["detail"] == ( "standby worker failed (RuntimeError): container did not start") +def test_cancelled_standby_emits_correlated_terminal_followup(tmp_path, monkeypatch): + from apps.web import run_manager as rm + from muteki.core.events import EventType + import apps.web.drivers as drivers + + entered = asyncio.Event() + + async def _waiting(run): + from muteki.core.events import Event, EventType + await run.bus.emit(Event( + event_type=EventType.FOLLOWUP_STARTED, + run_id=run.run_id, + payload={ + "followup_id": "followup-cancelled", "kind": "ask", + "question": "证据来源是什么？", + }, + )) + entered.set() + await asyncio.Future() + + monkeypatch.setattr( + drivers, "build_standby_driver", lambda cmd, mgr=None: _waiting, + ) + + async def _run(): + mgr = rm.RunManager(sessions_root=tmp_path / "sessions") + run = mgr.create("run-x") + run.started = True + run.finished = True + run.solved = True + assert mgr._ensure_standby(run.run_id, { + "action": "ask", "text": "证据来源是什么？", + "followup_id": "followup-cancelled", + }) + await entered.wait() + run.standby_task.cancel() + await asyncio.gather(run.standby_task, return_exceptions=True) + return [event async for event in run.store.replay(run.run_id)] + + events = asyncio.run(_run()) + lifecycle = [ + event for event in events if event.event_type in { + EventType.FOLLOWUP_STARTED, EventType.FOLLOWUP_FAILED, + } + ] + assert [event.event_type for event in lifecycle] == [ + EventType.FOLLOWUP_STARTED, EventType.FOLLOWUP_FAILED, + ] + assert {event.payload["followup_id"] for event in lifecycle} == { + "followup-cancelled", + } + assert lifecycle[-1].payload["detail"] == "后续操作已取消" + + def test_standby_final_cancel_log_redacts_callback_exception( tmp_path, monkeypatch, caplog): from apps.web import run_manager as rm @@ -749,9 +899,10 @@ async def _run(): assert raw_secret not in rendered -def test_resolve_reuses_challenge_from_winner_json(tmp_path, monkeypatch): - """resolve rebuilds the challenge from the durable winner.json snapshot so the - re-solve targets the same host.""" +def test_resolve_uses_private_challenge_and_ignores_workspace_winner( + tmp_path, monkeypatch, +): + """Worker-writable winner.json cannot redirect a resumed solve.""" from apps.web import run_manager as rm import apps.web.drivers as drivers @@ -769,11 +920,14 @@ async def _run(): mgr = rm.RunManager(sessions_root=str(tmp_path / "sessions")) run = mgr.create("run-x") run.started = True; run.finished = True; run.task = None - # drop a winner.json with the original challenge + mgr.persist_winner_continuation("run-x", {"challenge": { + "name": "expensey-eats", "category": "web", + "target": "https://target.example/"}}) + # A Worker can modify this compatibility artifact; it has no authority. wp = mgr.workspace_dir("run-x") / "winner.json" wp.write_text(json.dumps({"challenge": { - "name": "expensey-eats", "category": "web", - "target": "https://target.example/"}})) + "name": "tampered", "category": "pwn", + "target": "https://attacker.invalid/"}})) await mgr.resolve("run-x", {}) if run.task: await asyncio.gather(run.task, return_exceptions=True) diff --git a/tests/test_web_control.py b/tests/test_web_control.py index 87f9ba8..6186726 100644 --- a/tests/test_web_control.py +++ b/tests/test_web_control.py @@ -1049,6 +1049,20 @@ async def run(self): assert run.control_journal.context_bindings(context_id) == [ f"standby:{command_id}"] assert run.control_journal.context_delivery_status(context_id) == "bound" + + ask_command_id = "C-standby-secret-ask" + await mgr.post_control(run.run_id, { + "command_id": ask_command_id, + "action": "ask", + "target": "global", + "text": "password=ASK-SECRET", + }) + await run.control_actor.join() + ask_receipt = run.control_journal.latest_effect(ask_command_id) + assert ask_receipt is not None + assert ask_receipt.state is EffectState.EFFECT_OBSERVED + assert captured["hitl_cmd"]["text"] == "password=ASK-SECRET" + assert captured["hitl_cmd"]["followup_id"] await mgr.shutdown() diff --git a/tests/test_web_deck_ux.py b/tests/test_web_deck_ux.py index 280fd7d..a854902 100644 --- a/tests/test_web_deck_ux.py +++ b/tests/test_web_deck_ux.py @@ -413,6 +413,59 @@ def test_events_reducer_tracks_poc_blackboard_lifecycle(): _run_ui_node(script) +def test_followup_failure_replaces_its_correlated_pending_status(): + helper = UI_ROOT / "lib" / "events.ts" + script = textwrap.dedent( + f""" + const fs = require("fs"); + const ts = require("typescript"); + const vm = require("vm"); + const source = fs.readFileSync({json.dumps(str(helper))}, "utf8"); + const out = ts.transpileModule(source, {{ + compilerOptions: {{ module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 }} + }}).outputText; + const sandbox = {{ module: {{ exports: {{}} }}, exports: {{}} }}; + sandbox.exports = sandbox.module.exports; + vm.runInNewContext(out, sandbox, {{ filename: "events.js" }}); + const lib = sandbox.module.exports; + function assert(cond, msg) {{ if (!cond) throw new Error(msg); }} + + let s = lib.emptyDeck("run-followup"); + s = lib.reduce(s, {{ event_type: lib.EventType.FOLLOWUP_STARTED, + run_id: "run-followup", ts: 1, payload: {{ followup_id: "F1", + kind: "ask", question: "证据来源是什么？" }} }}); + s = lib.reduce(s, {{ event_type: lib.EventType.FOLLOWUP_STARTED, + run_id: "run-followup", ts: 1.1, payload: {{ followup_id: "F1", + kind: "ask", question: "证据来源是什么？" }} }}); + assert(s.followupPending, "follow-up is pending"); + assert(s.chat.filter((m) => m.content === "证据来源是什么？").length === 1, + "duplicate lifecycle event does not duplicate the question"); + assert(s.chat.some((m) => m.followupId === "F1" + && m.content === "正在回答追问…"), "pending row is correlated"); + + s = lib.reduce(s, {{ event_type: lib.EventType.FOLLOWUP_FAILED, + run_id: "run-followup", ts: 2, payload: {{ followup_id: "F1", + kind: "ask", detail: "后续操作已取消" }} }}); + assert(!s.followupPending, "terminal failure clears pending state"); + assert(!s.chat.some((m) => m.content === "正在回答追问…"), + "pending row is removed"); + assert(s.chat.filter((m) => m.followupId === "F1" + && m.content.includes("后续操作已取消")).length === 1, + "one correlated failure row remains"); + + s = lib.reduce(s, {{ event_type: lib.EventType.FOLLOWUP_STARTED, + run_id: "run-followup", ts: 3, payload: {{ followup_id: "F1", + kind: "ask", question: "证据来源是什么？" }} }}); + assert(!s.followupPending, + "a replayed start after terminal failure cannot restore pending state"); + assert(s.chat.filter((m) => m.followupId === "F1" + && m.content.includes("后续操作已取消")).length === 1, + "terminal row remains idempotent after replay"); + """ + ) + _run_ui_node(script) + + def test_events_reducer_shows_preflight_before_workers_and_failure_details(): helper = UI_ROOT / "lib" / "events.ts" script = textwrap.dedent( diff --git a/tests/test_worker_config.py b/tests/test_worker_config.py index 8b01c82..c55030f 100644 --- a/tests/test_worker_config.py +++ b/tests/test_worker_config.py @@ -194,6 +194,42 @@ def test_disabled_host_login_seat_can_remain_in_container_config(tmp_path) -> No assert config["engines"] == ["seat_codex_main"] +def test_atomic_configuration_validates_the_final_backend(tmp_path) -> None: + store = WorkerConfigStore(tmp_path) + seat = _seat("claude") + credential = _credential("claude", kind="system_inherit") + + config = store.set_configuration( + seats=[seat], + credentials=[credential], + worker_backend="local", + engines=[seat["id"]], + race_engines=[seat["id"]], + stage_policy={ + "coordinator": { + "review": {"enabled": False, "engine": seat["id"]}, + "verifier": {"enabled": False, "engine": ""}, + }, + }, + ) + + assert config["worker_backend"] == "local" + assert config["engines"] == [seat["id"]] + assert config["stage_policy"]["coordinator"]["review"]["engine"] == seat["id"] + + with pytest.raises(ValueError, match="系统登录"): + store.set_configuration( + seats=[seat], + credentials=[credential], + worker_backend="container", + engines=[seat["id"]], + ) + + reloaded = WorkerConfigStore(tmp_path).get() + assert reloaded["worker_backend"] == "local" + assert reloaded["credentials"] == [credential] + + def test_backend_network_and_budget_settings_persist(tmp_path, monkeypatch) -> None: monkeypatch.setattr("apps.web.worker_config.is_web_container", lambda: False) WorkerConfigStore(tmp_path).set( From 054e64ae94749404bdd47fdd983e736a7a0b73c3 Mon Sep 17 00:00:00 2001 From: jiayuqi7813 <1783671926@qq.com> Date: Sat, 22 Aug 2026 21:29:09 +0800 Subject: [PATCH 2/2] fix(web): recover interrupted follow-ups --- apps/web/run_manager.py | 100 ++++++++++++++++++++++++++++++++++++- tests/test_standby_hitl.py | 50 +++++++++++++++++++ tests/test_web_server.py | 69 +++++++++++++++++++++++++ 3 files changed, 218 insertions(+), 1 deletion(-) diff --git a/apps/web/run_manager.py b/apps/web/run_manager.py index 71f5b52..c09d540 100644 --- a/apps/web/run_manager.py +++ b/apps/web/run_manager.py @@ -428,6 +428,8 @@ def __init__(self, *, sessions_root: "str | Path | None" = None, # dispatch path falls back to this when a request doesn't say otherwise. self.worker_config = WorkerConfigStore(root=self.sessions_root) protocol2_run_ids = self._reconcile_protocol2_flags() + self._recover_interrupted_followups( + SessionStore(root=self.sessions_root)) self._rehydrate(protocol2_run_ids=protocol2_run_ids) def _execution_owned( @@ -747,6 +749,80 @@ def _reconcile_protocol2_flags(self) -> Optional[frozenset[str]]: ) return frozenset(catalog_run_ids) + def _recover_interrupted_followups(self, store: SessionStore) -> None: + """Persist a terminal event for follow-ups abandoned by a process exit. + + Ask and writeup executions belong to the web process that started them. + During manager construction there cannot be a surviving standby task from + the previous process, so every durable ``followup.started`` without a + matching terminal event is interrupted. Recording the recovery in the + same JSONL log keeps replay deterministic and unlocks every fresh client, + rather than synthesizing a different result per SSE connection. + """ + started_type = EventType.FOLLOWUP_STARTED.value + terminal_types = { + EventType.FOLLOWUP_COMPLETED.value, + EventType.FOLLOWUP_FAILED.value, + } + for run_id in store.list_runs(): + pending: dict[str, dict[str, Any]] = {} + try: + for index, row in enumerate(store.load_all(run_id)): + event_type = str(row.get("event_type") or "") + payload = row.get("payload") + payload = payload if isinstance(payload, dict) else {} + followup_id = str(payload.get("followup_id") or "") + if event_type == started_type: + key = followup_id or f"legacy:{index}" + pending[key] = { + "followup_id": followup_id, + "kind": str(payload.get("kind") or "ask"), + "execution_generation": payload.get( + "execution_generation"), + "recovery_id": ( + f"interrupted-followup:{index}:{followup_id}" + ), + } + elif event_type in terminal_types: + if followup_id: + pending.pop(followup_id, None) + else: + legacy_key = next( + (key for key in reversed(pending) + if not pending[key]["followup_id"]), + None, + ) + if legacy_key is not None: + pending.pop(legacy_key, None) + + for interrupted in pending.values(): + payload = { + "followup_id": interrupted["followup_id"], + "kind": interrupted["kind"], + "detail": "服务已重启，后续操作已中断", + "recovery_id": interrupted["recovery_id"], + } + generation = interrupted.get("execution_generation") + if generation is not None: + payload["execution_generation"] = generation + recovery = Event( + event_type=EventType.FOLLOWUP_FAILED, + run_id=run_id, + solver_id="web-runtime-recovery", + seq=store.last_stream_seq(run_id) + 1, + payload=payload, + ) + store.append_if_absent_sync( + recovery, + identity_field="recovery_id", + identity=interrupted["recovery_id"], + ) + except Exception as exc: + LOG.error( + "Interrupted follow-up recovery failed for %s error_type=%s", + run_id, type(exc).__name__, + ) + def _rehydrate( self, *, protocol2_run_ids: Optional[frozenset[str]] = None ) -> None: @@ -3161,10 +3237,27 @@ def _ensure_standby(self, run_id: str, cmd: dict[str, Any]) -> bool: driver = build_standby_driver(cmd, mgr=self) async def _go() -> None: + followup_terminal = False + + async def _note_followup_terminal(ev: Event) -> None: + nonlocal followup_terminal + if ev.event_type not in { + EventType.FOLLOWUP_COMPLETED, + EventType.FOLLOWUP_FAILED, + }: + return + wanted = str(cmd.get("followup_id") or "") + event_id = str(ev.payload.get("followup_id") or "") + if not wanted or event_id == wanted: + followup_terminal = True + async def _emit_followup_failed(detail: str) -> None: - if action not in {"ask", "writeup"}: + nonlocal followup_terminal + if action not in {"ask", "writeup"} or followup_terminal: return try: + if bool(getattr(run.bus, "_closed", False)): + self._fresh_bus(run) await run.bus.emit(Event( event_type=EventType.FOLLOWUP_FAILED, run_id=run_id, @@ -3174,9 +3267,11 @@ async def _emit_followup_failed(detail: str) -> None: "detail": detail, }, )) + followup_terminal = True except Exception: pass + run.bus.add_sink(_note_followup_terminal) try: LOG.info("standby worker starting for %s action=%s", run_id, cmd.get("action")) @@ -3210,6 +3305,9 @@ async def _emit_followup_failed(detail: str) -> None: except Exception: pass finally: + run.bus.remove_sink(_note_followup_terminal) + if action in {"ask", "writeup"} and not followup_terminal: + await _emit_followup_failed("后续操作已中断") # Do not close the bus; retain the completed task as an observable # receipt. `_ensure_standby` checks `.done()` and replaces it on the # next command, so this does not block subsequent follow-ups. diff --git a/tests/test_standby_hitl.py b/tests/test_standby_hitl.py index eba885b..0cffc1d 100644 --- a/tests/test_standby_hitl.py +++ b/tests/test_standby_hitl.py @@ -869,6 +869,56 @@ async def _run(): assert lifecycle[-1].payload["detail"] == "后续操作已取消" +def test_standby_wrapper_fails_followup_when_driver_exits_without_terminal( + tmp_path, monkeypatch): + from apps.web import run_manager as rm + from muteki.core.events import Event, EventType + import apps.web.drivers as drivers + + async def _started_then_exit(run): + await run.bus.emit(Event( + event_type=EventType.FOLLOWUP_STARTED, + run_id=run.run_id, + payload={ + "followup_id": "followup-orphan", + "kind": "ask", + "question": "证据来源是什么？", + }, + )) + + monkeypatch.setattr( + drivers, "build_standby_driver", lambda cmd, mgr=None: _started_then_exit, + ) + + async def _run(): + mgr = rm.RunManager(sessions_root=tmp_path / "sessions") + run = mgr.create("run-x") + run.started = True + run.finished = True + run.solved = True + assert mgr._ensure_standby(run.run_id, { + "action": "ask", + "text": "证据来源是什么？", + "followup_id": "followup-orphan", + }) + await asyncio.gather(run.standby_task, return_exceptions=True) + return [event async for event in run.store.replay(run.run_id)] + + events = asyncio.run(_run()) + lifecycle = [ + event for event in events if event.event_type in { + EventType.FOLLOWUP_STARTED, + EventType.FOLLOWUP_FAILED, + } + ] + assert [event.event_type for event in lifecycle] == [ + EventType.FOLLOWUP_STARTED, + EventType.FOLLOWUP_FAILED, + ] + assert lifecycle[-1].payload["followup_id"] == "followup-orphan" + assert lifecycle[-1].payload["detail"] == "后续操作已中断" + + def test_standby_final_cancel_log_redacts_callback_exception( tmp_path, monkeypatch, caplog): from apps.web import run_manager as rm diff --git a/tests/test_web_server.py b/tests/test_web_server.py index 935ca68..67e0b18 100644 --- a/tests/test_web_server.py +++ b/tests/test_web_server.py @@ -938,6 +938,75 @@ async def seed() -> None: assert r2.finished is True # force-settled (was a ghost otherwise) +def test_rehydrate_persists_failure_for_interrupted_followup(tmp_path) -> None: + from muteki.core.events import Event + from muteki.core.session_store import SessionStore + + sessions = tmp_path / "sessions" + mgr1 = RunManager(sessions_root=sessions) + run = mgr1.create("interrupted-followup") + + async def seed() -> None: + await run.bus.emit(Event( + event_type=EventType.RUN_STARTED, + run_id=run.run_id, + payload={"challenge": {"name": "x"}}, + )) + await run.bus.emit(Event( + event_type=EventType.RUN_FINISHED, + run_id=run.run_id, + payload={"solved": True}, + )) + await run.bus.emit(Event( + event_type=EventType.FOLLOWUP_STARTED, + run_id=run.run_id, + payload={ + "followup_id": "F-complete", + "kind": "ask", + "question": "已经回答的问题", + }, + )) + await run.bus.emit(Event( + event_type=EventType.FOLLOWUP_COMPLETED, + run_id=run.run_id, + payload={ + "followup_id": "F-complete", + "kind": "ask", + "text": "回答", + }, + )) + await run.bus.emit(Event( + event_type=EventType.FOLLOWUP_STARTED, + run_id=run.run_id, + payload={ + "followup_id": "F-interrupted", + "kind": "writeup", + }, + )) + + asyncio.run(seed()) + + RunManager(sessions_root=sessions) + events = SessionStore(sessions).load_all(run.run_id) + recovered = [ + event for event in events + if event["event_type"] == EventType.FOLLOWUP_FAILED.value + ] + assert len(recovered) == 1 + assert recovered[0]["payload"]["followup_id"] == "F-interrupted" + assert recovered[0]["payload"]["kind"] == "writeup" + assert recovered[0]["payload"]["detail"] == "服务已重启，后续操作已中断" + assert recovered[0]["solver_id"] == "web-runtime-recovery" + + # Recovery is durable and idempotent across subsequent restarts. + RunManager(sessions_root=sessions) + events = SessionStore(sessions).load_all(run.run_id) + assert sum( + event["event_type"] == EventType.FOLLOWUP_FAILED.value + for event in events + ) == 1 + + def test_rehydrate_protocol2_started_run_stays_unfinished(tmp_path) -> None: from muteki.core.events import Event