diff --git a/docs/architecture/01-high-level-architecture.md b/docs/architecture/01-high-level-architecture.md index 338aa8182..6e9a3fb7a 100644 --- a/docs/architecture/01-high-level-architecture.md +++ b/docs/architecture/01-high-level-architecture.md @@ -27,7 +27,7 @@ server.py (Starlette+uvicorn) ← HTTP + WebSocket on configurable host:port (de │ ├── modules/project_work_pointer.js, project_read_state.js ← Pointer to a loaded root card, no execution or history authority; a room's read receipt (§3 Project rooms) │ ├── modules/model_wait.js ← Model-wait views and owner actions inside existing chat cards, through the shared decision ingress (§6 Quota and auth waits) │ ├── modules/task_checkpoints.js, cancel_presentation.js ← Typed task checkpoints and recorded cancellation-cause text shared by Chat and Logs (§3) - │ ├── modules/dashboard.js, logs.js, costs.js, files.js ← Dashboard tab host; Logs (backfill plus live-stream duplicate guard); Costs (an open zero is never shown as free); Files browser over `/api/files/*` (§3 Dashboard, Files) + │ ├── modules/dashboard.js, logs.js, worker_sha_presentation.js, costs.js, files.js ← Dashboard host; Logs (backfill/live dedupe, factual checkout ancestry); Costs (unknown is not free); Files over `/api/files/*` (§3) │ ├── modules/skills.js, marketplace.js, skill_review_card.js, skill_publish_flow.js ← Installed-skills UI; the ClawHub marketplace inside Skills; Skill Review chat cards; publish-dialog detail rows (§3 Skills and Widgets) │ ├── modules/settings_ui.js, settings_catalog.js, settings_controls.js, settings_secrets.js, settings_local_model.js, settings_autostart.js, mcp_settings.js ← Settings page (Accounts → Secrets → Models → Agents); model-catalog refresh with a 25-second bound and a sequence guard; control binders; shared full-secret Show/Hide; the local-model form; the host sign-in toggle; MCP cards that keep masked tokens until edited (§3 Settings and onboarding) │ ├── modules/model_roles.js, model_chooser.js ← The Models editor shared by Settings and onboarding; the editable chooser shared with the route editors — catalog arrival never assigns a value (§3 Navigation and shared UI contracts) @@ -41,7 +41,7 @@ server.py (Starlette+uvicorn) ← HTTP + WebSocket on configurable host:port (de │ ├── active_activity.py ← process-local owner of in-flight native chat actors (`DirectActivityRegistry`): private handles for controls and the writer drain, public snapshots for `/api/state` `active_direct_turns` and WS typing frames; no queue records (§3 Direct turns and the activity block) │ ├── message_bus.py ← Queue-based local message bus (Web UI + reviewed transport skills) │ ├── workers.py ← Multiprocessing worker pool (forkserver on Linux, spawn on macOS/Windows; never fork from the multi-threaded supervisor) - │ ├── worker_assignment.py, worker_chat_lane.py, worker_health.py, worker_pool_lifecycle.py, worker_process.py, worker_promotion.py ← The pool's leaves: assignment and its refusals; the direct chat lane (restart resume; admission during a managed update — DEVELOPMENT "Managed Update Rule"); health-owned crash detection and terminal-file recovery; pool lifecycle and `kill_worker_tree`, the ONE worker tree-kill (daemon roots always spared); the worker child process; chat-turn/project-scope promotion to a queued task (single refusal writer `_persist_promote_rejection`) (§5; §6 Owner routing verbs) + │ ├── worker_assignment.py, worker_chat_lane.py, worker_health.py, worker_pool_lifecycle.py, worker_process.py, worker_promotion.py ← Pool leaves: assignment/refusals; direct chat/restart/update admission; crash and terminal-file recovery; lifecycle and `kill_worker_tree` (daemon roots spared), diagnostic `worker_sha_verify` ancestry (normal descendants only in Logs); worker child; chat/project promotion via `_persist_promote_rejection` (§5; §6 Owner routing verbs) │ ├── worker_owner_wait.py ← Queue-owned active-capacity transfer for a required owner wait: the task stays RUNNING and custodied while another worker takes the active slot (§5) │ ├── state.py, state_initialization.py ← `state/state.json` field updates, typed read quality; init witness │ ├── queue.py ← Task queue (PENDING/RUNNING) + activity-based timeouts; the ONE task-state authority, extended by the lifecycle/publication/transition modules below (§5) @@ -77,7 +77,7 @@ server.py (Starlette+uvicorn) ← HTTP + WebSocket on configurable host:port (de │ ├── update_source.py ← Official update-source selection + network policy through the bounded Git runner │ ├── update_recovery.py ← Exact owner Restore/promotion: pinned prior HEAD, rescue-before-reset, one captured SHA for local/remote promotion │ ├── update_merge.py ← Managed-update engine: exact-target 3-way plan, clean fast-forward, stash-first reviewed assisted merge, the transaction (`m0_tree`, `tests_evidence`, `stash_sha`/`local_work_carrier`, `failed_update_ref`), verified rollback/smoke, boot recovery; residual: M0 is a pin-once forensic baseline in the resolver-writable tx marker, disclosed by review, never re-verified (DEVELOPMENT "Managed Update Rule") - │ ├── update_candidate.py ← Candidate/carrier primitives: private-index tree serialization, rerere-neutral merges (no silent rr-cache replay), failed-update preservation branches, marker-guarded stash restore; `tests_evidence` is forensics, never reuse authority (§6 Hermetic preflight proof) + │ ├── update_candidate.py ← Candidate/carrier primitives: private-index tree serialization, rerere-neutral merges (no silent rr-cache replay), failed-update preservation branches, write-ahead stash restore, retained backups and exact conflict diagnostics; `tests_evidence` is forensics, never reuse authority (§6 Hermetic preflight proof) │ ├── update_carriers.py, update_merge_plan.py ← Carrier-aware conflict resolution (the version/contract carriers a merge may not resolve by hand), and merge planning with live materialization │ └── update_merge_policy.py ← Presentation-only doc/code/hot conflict labels; every conflict uses the same reviewed assisted path │ @@ -364,7 +364,7 @@ server.py (Starlette+uvicorn) ← HTTP + WebSocket on configurable host:port (de ├── server_entrypoint.py ← CLI parsing + port binding helpers ├── server_runtime.py ← Startup/onboarding wiring + WS liveness ├── server_web.py ← Static/web roots and fixed-source `read_author_kit_assets(repo_dir)`; no endpoint/cache - ├── server_process.py, server_liveness.py, server_maintenance.py, server_restart.py, server_owner_routing.py, server_routing_context.py ← Server leaves the composition root calls: the facts one server process shares with every leaf, including applied startup settings; wedge detection for the supervisor generation; the upkeep a generation owes the drive; restart operations (shutdown, the checkout-first manual Restart, the planned restart's engine-pin daemon stop; §9); where one owner message goes; the bounded facts one owner turn may address + ├── server_process.py, server_liveness.py, server_maintenance.py, server_restart.py, server_owner_routing.py, server_routing_context.py ← Server leaves the composition root calls: the facts one server process shares with every leaf, including applied startup settings and the launcher stop pipe (§9); wedge detection for the supervisor generation; the upkeep a generation owes the drive; restart operations (shutdown, the checkout-first manual Restart, the planned restart's engine-pin daemon stop; §9); where one owner message goes; the bounded facts one owner turn may address ├── terminal_cost_reconciliation.py ← Usage recovery/projection (§6) ├── task_continuation.py ← Durable review continuation state ├── task_results.py ← Durable task results `task_results/.json`; the locked `task_acceptance_review_accounting` claim (minted at first physical reviewer dispatch; a claim without a recoverable terminal host run is UNKNOWN, never permission to re-dispatch); the read-only root review-capacity projection is WALLET and cancellation only (`root_task_id`, `cap_cycles`, `claimed_cycles`, `remaining_cycles`, `binding_seen`, `dedupe`, `state`, `reason`), no time axis (§6 Task acceptance) diff --git a/docs/architecture/05-supervisor-loop.md b/docs/architecture/05-supervisor-loop.md index adafe7bb9..192a42439 100644 --- a/docs/architecture/05-supervisor-loop.md +++ b/docs/architecture/05-supervisor-loop.md @@ -67,7 +67,7 @@ The event bus is process-lifetime, not worker-generation-lifetime: respawns reus Heartbeat and progress differ: a heartbeat proves a process/loop alive; owner-visible progress and model-usage events prove the task advanced. Progress keeps an orchestrator alive, but deadline, Stop and budget stay hard. The absolute ceiling ends solve/owner-stop finalization, not settled post-work; Stop, deadline, money and per-call bounds still apply (§6). After the typed finalization episode, timeout handling freezes its decision under Q, marks the worker `reaping`, and hands kill, join, salvage, retry and respawn to the single off-loop reaper; an orchestrator with live descendants is never blindly retried (a retry would replay its plan, spawning a competing tree). No retry or assignment may occupy a timed-out slot until the original process is provably dead: if kill and join cannot establish death, the reaper keeps a low-rank RUNNING result and the `reaping` slot, emits a visible `task_reaper_wedged` receipt and restart hint, and writes no terminal, `task_done`, retry or respawn: one slot is sacrificed rather than letting a live process race a replacement; the next supervisor generation reconciles the record after old-generation custody. Typed timeout codes (`queue_timeouts.TIMEOUT_TERMINAL_REASONS`) keep their `task_incident` identity; owner-facing grace/kill/salvage text uses `project_dialogue.TASK_CAUSE_PHRASES`, unknown codes raw. -A spawned/respawned slot is unassignable until its child's PID-bound `worker_ready` row arrives (`supervisor/worker_pool_lifecycle.py`). A live child's own `worker_starting` row (before extension loading and agent construction) permits one extension of `WORKER_READY_WINDOW_SEC` to `WORKER_READY_CEILING_SEC` (300 s from birth; both in `runtime_limits.py`); foreign or pre-spawn rows cannot extend another slot; `worker_ready_window_extended` records the decision. A silent child keeps its original window; logging failure cannot block startup. After `WORKER_READY_MAX_ATTEMPTS` failed attempts, `Worker.readiness_exhausted` is final for that exact slot: late events cannot reopen it. Total exhaustion (distinct from busy/booting/reaping capacity and live owner-wait stacks) closes pooled ingress (owner `/review` included) without blocking direct chat/control or boot/update recovery; once RUNNING completion custody has settled, `disable_exhausted_worker_pool` fails unstarted PENDING work honestly with a Restart hint, and a new task cannot clear the latch. Readiness is separate from liveness and task idle time; a watcher error releases only still-booting, non-exhausted slots to the crash detector (`worker_ready_released`). Linux workers use forkserver; macOS/Windows use spawn. +A new/respawned slot opens on its child’s PID-bound `worker_ready` (`supervisor/worker_pool_lifecycle.py`), `worker_sha_verify.ok` keeps equality; ancestry is diagnostic. Descendants stay in Logs; non-descendants and unavailable applicable comparisons warn Main; missing baseline stays skipped. Neither attests loaded code/review. A live child’s own `worker_starting` (before extensions/agent construction) permits one extension from `WORKER_READY_WINDOW_SEC` to `WORKER_READY_CEILING_SEC` (300 s from birth; `runtime_limits.py`), recorded as `worker_ready_window_extended`. Foreign/pre-spawn rows cannot extend it; silence keeps the original window and logging failure cannot block startup. After `WORKER_READY_MAX_ATTEMPTS`, `Worker.readiness_exhausted` is final for that slot; late events cannot reopen it. Total exhaustion closes pooled ingress, including owner `/review`, but not direct chat/control or boot/update recovery. After RUNNING custody settles, `disable_exhausted_worker_pool` fails unstarted PENDING rows with a Restart hint; new tasks cannot clear the latch. Readiness differs from liveness/task idle time. Watcher error releases only still-booting, non-exhausted slots to crash detection (`worker_ready_released`). Linux uses forkserver; macOS/Windows use spawn. `worker_health.recover_confirmed_dead_worker` reserves exact custody under Q, then enqueues reaper `confirmed_dead_worker`. Saved terminal bytes win even after signal death; unknown/incomplete publication retains `TerminalFileRecoveryPending`. Only confirmed absence reaches crash policy: signal means infrastructure failure; eligible non-signal crashes retry within `QUEUE_MAX_RETRIES`, preserving cost and owner-wait restrictions. Budget continuation forbids ordinary retry; unconsumed grants re-park, consumed grants remain terminal, and failed revocation retains source/grant with honest persistence status (§6). diff --git a/docs/architecture/09-shutdown-and-process-cleanup.md b/docs/architecture/09-shutdown-and-process-cleanup.md index 6f37a0c89..48f134346 100644 --- a/docs/architecture/09-shutdown-and-process-cleanup.md +++ b/docs/architecture/09-shutdown-and-process-cleanup.md @@ -1,12 +1,12 @@ # 9. Shutdown & Process Cleanup -Window close retains the shared daemon; manual Restart stops owned work then re-execs; Panic explicitly stops. Signals, uncertain custody and next-boot reconciliation never imply unproven death. +Close retains the shared daemon; Restart stops owned work then re-execs; Panic stops. Signals and recovery never prove process death. **Close, quit, Panic (desktop).** `launcher_background.py` decides a close of the desktop window from `OUROBOROS_DESKTOP_KEEP_RUNNING`, read at that moment. On (Windows, macOS) it hides the window while an indicator is live and everything keeps running: tasks, schedules, consciousness, Telegram and Presence companions, the shared daemon. Off, or with no live indicator, it quits through the ordinary window-close teardown below. Never decided, it keeps this close, asks once from a worker thread (pywebview's Cocoa confirmation deadlocks inside `closing`), stores the answer and quits on dismissal. A quit (the indicator's Quit, Cmd+Q, Dock Quit, sign-out, shutdown) is never asked or cancelled: Windows decides in the form's own FormClosing, the only place carrying CloseReason, and macOS marks `applicationShouldTerminate:` before pywebview runs its shared `closing` event. Quit sets no Panic flag and leaves sign-in startup alone. Panic keeps its own path: the server exits with the Panic code, and the lifecycle thread asks the indicator to go without waiting for it before orphan cleanup and `os._exit`; the indicator's Panic item posts `/api/command /panic` over loopback HTTP, as Android's notification does. A hidden window returns when its indicator dies, or the launcher exits when the window cannot be shown. A manual second launch shows the running window (Windows: a named auto-reset kernel event derived from the PID-lock path, not a durable file; POSIX: SIGURG to the PID in the lock file, which an older launcher ignores; a lock loser no longer truncates that file), also while that copy still boots: it listens right after taking the lock, and a request that comes before pywebview has shown the window is kept until then and cancels a quiet start; an automatic second launch only logs. Linux keeps the plain close. Graceful/forced window close stops this generation's workers/services, waits for its group/Job, cleans owned processes and releases the PID lock, preserving the shared daemon and other clients. Windows keeps the daemon outside the launcher Job; old immutable launchers need a package update. -`launcher.stop_agent` terminates only the server PID on every platform, waits `LAUNCHER_STOP_GRACE_SEC`, then uses group SIGKILL and the post-exit sweep. Early group SIGTERM would destroy the server's Manager/pool before teardown, causing `EVENT_Q` BrokenPipe and false `supervisor_failure`. For old launchers, `server_process._SignalStopServer.handle_exit` sets `_supervisor_stop` on signal, so shutdown errors are not crashes. Uvicorn uses `SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_SEC` within the launcher budget (`runtime_limits.py`); embedded Host Service sets `capture_signals=nullcontext` on `_embedded_uvicorn_server` so its drain never owns process signals. The permanent `_exit_signalled` latch refuses `_start_supervisor_if_needed`, ends a just-admitted `_supervisor_generation` and fences loop/crash paths: settings cannot revive teardown. `_supervisor_stop` remains revivable for crash recovery/fresh lifespan (§5). `tests/test_shutdown_signal_1142.py` covers old group SIGTERM with an open request. +`launcher.stop_agent` sends POSIX SIGTERM only to the server, its Manager/pool owner. Windows sends `quit\n` through the child’s private stdin (`OUROBOROS_LAUNCHER_STOP_STDIN=1`); `_SignalStopServer.watch_launcher_stop` consumes the opt-in before workers, sets stop/exit latches and ends Uvicorn. EOF is not Quit; direct-server stdin is untouched. Both paths keep `LAUNCHER_STOP_GRACE_SEC` and forced group/Job cleanup. Old Windows launchers force `terminate()` until updated. Old group SIGTERM kills the Manager early: `handle_exit` sets `_supervisor_stop` immediately so BrokenPipe is not a supervisor crash. Uvicorn’s `SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_SEC` leaves custody time; embedded Host Service uses instance `capture_signals=nullcontext` for one signal owner. `_exit_signalled` prevents settings or admitted generations reviving teardown; `_supervisor_stop` remains revivable (§5). `tests/test_shutdown_signal_1142.py` covers group signal and pipe with open connections. `server_shutdown.cause` distinguishes `external_signal` and `launcher_quit`. Planned restart retains same-pin daemons. After worker custody, before optional cleanup, graceful and final paths call `_stop_owned_daemon_for_new_pin`: `load_runtime_pin` versus attach-only `read_owned_gateway`. Unpublished/unreadable pins and unreachable/foreign/stopped daemons leave handoff unchanged; unconfirmed stop retains custody/diagnostics for reconciliation without invented spend. diff --git a/docs/development/07-managed-update-rule.md b/docs/development/07-managed-update-rule.md index 9a659d1c5..dfdbaa0aa 100644 --- a/docs/development/07-managed-update-rule.md +++ b/docs/development/07-managed-update-rule.md @@ -1,16 +1,18 @@ # Managed Update Rule -This chapter owns what a managed update may do to the live tree: how one exact official target is chosen and bound, which windows refuse conversation, how dirty local work is stashed rather than merged into history, what the authorized resolver may stage, and when a fresh rescue must precede destruction. It exists because a managed update rewrites the live tree in place under owner authority while other work may be running, so every step of it needs a stated custody rule. +Managed update binds targets and owns writer admission, local-work custody, resolver authority and recovery of the live tree. -- Keep the local work branch and the official update feed separate; the channel and branch topology live in ARCHITECTURE "8. Git Branching, CI, and Build" (`ouroboros/update_channels.py`). -- A preflight chooses one exact official target SHA. Apply binds to the disclosed base/target, closes new writers, drains direct turns, stops workers and tracked services, then re-plans before mutation; the update transaction is written before mutation, and writers reopen only after a verified abort/rollback or a healthy restart. Delayed evolution cleanup takes the same update lock and admission owner. -- Conversation admission is separate from repo-writing permission: the chat lanes refuse turns only during destructive windows (apply/replace/rollback prologue, materialization). While the ONE authorized assisted resolver holds the repository (`assisted_resolution` / `committing_assisted`), Main keeps answering, the registry guard still refuses repo tools to every other task, and steering reaches the resolver through the ordinary `steer_task` mailbox (`supervisor/worker_chat_lane.py::conversation_admitted_during_update`). The server's first-party packages are imported BEFORE conflict markers land in the live tree (`preload_owner_control_path`), because the packaged server runs embedded Python from the materialized repository. -- Dirty local work never enters merge history: the apply stashes it and restores it as uncommitted content; a conflicting restore keeps the stash and discloses the recovery command. The reviewed assisted resolver runs only when Git reports a real conflict — filenames do not create a second update policy. Managed materialization and rollback run their internal `git reset --hard` without interactive confirmation; the explicit-confirmation rule applies to the owner-facing generic restore seam. -- The authorized resolver stages the complete merge including tracked binary files, and review receives their exact staged mode/blob/size plus the parent object ids; missing exact metadata blocks. This exception does not weaken the ordinary commit pipeline's binary policy. -- A managed merge commits only with proof that the full suite ran green on the exact candidate tree ("The commit gate mirrors the CI split" names the proof authority). Any non-commit terminal of the resolver rolls the live tree back and best-effort preserves the attempt on the deterministic failed-update branch; the fresh rescue snapshot, not that branch, is what rollback depends on. -- Take a fresh rescue before every destructive rollback and before boot-resume re-materialization — fail-open but disclosed durably at capture time, with its pointer in the update transaction (the choke point and why the pre-update snapshot is not enough: ARCHITECTURE "2. Startup / Onboarding Flow"). -- Manual Restore reuses the same writer fence and pins the previous HEAD on a local recovery branch before reset. Promotion resolves the development SHA once and uses that exact SHA for both the local QA ref and any remote push. +- Local work branch and official feed stay separate; topology: ARCHITECTURE §8 (`ouroboros/update_channels.py`). +- Preflight selects one exact target. Apply binds base/target, closes new writers, drains direct turns, stops workers/services and re-plans before mutation. Write the transaction first; reopen only after verified abort/rollback or healthy restart. Delayed evolution cleanup uses the same update lock/admission owner. +- Conversation and repo-writing differ. Destructive apply/replace/rollback/materialization refuses turns; while the authorized assisted resolver owns `assisted_resolution`/`committing_assisted`, Main answers but other tasks’ repo tools remain refused. Steering uses ordinary `steer_task`; admission lives in `worker_chat_lane.conversation_admitted_during_update`. `preload_owner_control_path` imports server packages before conflict markers enter the materialized checkout. +- Dirty work never enters merge history: stash it, then restore it uncommitted. Never automatically drop update stashes, even after success. `rescue-local-*` pins the exact object; explicit Git cleanup remains available. Only a real Git conflict invokes the reviewed resolver; filenames create no policy. Internal materialization/rollback reset without interactive confirmation; the generic owner-facing Restore retains its confirmation. +- Record restoration intent in the existing transaction BEFORE apply. Each completion path consumes restored/preserved/incomplete outcomes; incomplete persistence retains the marker. Interrupted replay neither resets nor reapplies over returned or later work: once storage works, record a handoff of the pinned backup, preserving current files without claiming clean/successful apply. Failure captures rc/stdout/stderr and NUL-separated unmerged paths before verified cleanup; release carries one recovery note to Main/Updates and supervisor events. Unreadable census differs from empty; neither counts successful files or enumerates untracked collisions. +- Ordinary transactions keep schema 1; a `stash_restore` record requires schema 2. New readers accept legacy markers; older strict readers refuse new semantics without reset. Rollback checks out old code before restoration, so interrupted new records can require a capable reader or deliberate recovery: refusal is not automatic recovery. Restart the actual desktop app, Docker/service or direct server, not its browser tab. In-app Restart refuses pending restoration before checkout/reset. Legacy `stash_restored` still means recorded successful apply. +- The resolver stages the full merge, tracked binaries included; review gets exact staged mode/blob/size and parent object IDs. Missing metadata blocks; ordinary commit binary policy is unchanged. +- Managed merge commits require a green full suite on the exact candidate (DEVELOPMENT “The commit gate mirrors the CI split”). Any non-commit resolver terminal rolls back and best-effort preserves the attempt on its deterministic failed-update branch. Recovery needs the fresh rescue, not that branch. +- Fresh rescue precedes every destructive rollback and boot re-materialization. Capture failure is fail-open but durably disclosed, with a transaction pointer; ARCHITECTURE §2 owns the choke point and why pre-update rescue is insufficient. +- Manual Restore uses the same writer fence and pins previous HEAD before reset. Promotion resolves development SHA once for both local QA and remote push. -Enforcement: `tests/test_update_merge_policy.py` (what the merge policy refuses), `tests/test_update_dirty_stash.py` (the dirty-tree path), `tests/test_update_hardening.py` and `tests/test_update_tx_corrupt_quarantine.py` (transaction integrity and quarantine). +Tests: `tests/test_update_merge_policy.py`, `test_update_dirty_stash.py`, `test_update_stash_recovery.py` (real Git, crash/replay, consumers, Restart), `test_update_hardening.py`, `test_update_tx_corrupt_quarantine.py`. -External native hosts add no source updater: the source-selected launcher hook installs/verifies the artifact before a core generation, and managed post-boot readback precedes finalization. Failure uses existing rollback; worker self-restart keeps incomplete native adoption even when core SHA matches. +Native hosts add no source updater: the launcher hook verifies its installed artifact before core start; post-boot readback precedes finalization. Failure uses existing rollback. Worker self-restart retains incomplete native adoption even when core SHA matches. diff --git a/launcher.py b/launcher.py index 56d553417..b8d5d4611 100644 --- a/launcher.py +++ b/launcher.py @@ -19,13 +19,9 @@ import webbrowser from logging.handlers import RotatingFileHandler from typing import Optional -# WA6: set sys.dont_write_bytecode BEFORE importing any project module. A signed -# macOS .app must never write __pycache__/*.pyc into its own bundle at runtime -# (that breaks the codesign seal and triggers AppTranslocation). os.environ alone -# is INSUFFICIENT for THIS process: PYTHONDONTWRITEBYTECODE is only read at -# interpreter startup, so mutating os.environ later does not stop the current -# process's own subsequent imports — only sys.dont_write_bytecode does. The env -# vars set further below propagate the same policy to child processes. +# Seal-safe imports: the current interpreter needs the flag before project imports; +# PYTHONDONTWRITEBYTECODE alone only affects later child interpreters. Bundle writes +# would invalidate a macOS signature and trigger AppTranslocation. sys.dont_write_bytecode = True os.environ.setdefault("PYTHONDONTWRITEBYTECODE", "1") @@ -396,14 +392,8 @@ def start_agent(port: int = AGENT_SERVER_PORT) -> subprocess.Popen: env["OUROBOROS_APP_VERSION"] = str(APP_VERSION) env["OUROBOROS_MANAGED_BY_LAUNCHER"] = "1" env["OUROBOROS_MANAGED_REPO_DIR"] = str(REPO_DIR.resolve()) - # Owner Surface Fact: the launcher alone knows presentation; `_headless` is decided in main() before the - # lifecycle loop ever calls start_agent(), and every managed restart funnels - # back through here, so the export is re-stamped fresh each time. Absence of - # the var (source mode, Docker, Colab, CLI server) truthfully means "web". - # Env-only by design — never a SETTINGS_DEFAULTS key (pop-on-absent would - # erase an injected value). Known bounded lie: a SIGKILLed launcher can - # orphan the server with a stale "desktop_window" until the next launcher - # start reaps it — the same envelope OUROBOROS_MANAGED_BY_LAUNCHER accepts. + # Launcher-owned presentation is re-stamped for every generation; other hosts default + # to web. After a killed launcher it can remain stale until the next owned reap. env["OUROBOROS_PRESENTATION"] = ( str(os.environ.get("OUROBOROS_PRESENTATION") or "web") if _external_ui else "browser_fallback" if _headless else "desktop_window" @@ -415,9 +405,7 @@ def start_agent(port: int = AGENT_SERVER_PORT) -> subprocess.Popen: else: env.pop("OUROBOROS_EXTERNAL_HOST_UPDATE", None) env.pop("OUROBOROS_EXTERNAL_HOST_RESULT", None) - # The server runs out of the managed repo, not the bundle: without this the - # bundled payloads (node, ripgrep) are invisible to it (platform_layer. - # bundled_resource_bases). + # The managed server still resolves payloads from this bundle. env[BUNDLE_DIR_ENV] = str(_bundle_dir()) server_py = REPO_DIR / "server.py" @@ -430,6 +418,8 @@ def start_agent(port: int = AGENT_SERVER_PORT) -> subprocess.Popen: "stderr": subprocess.STDOUT, } if IS_WINDOWS: + env["OUROBOROS_LAUNCHER_STOP_STDIN"] = "1" + popen_kwargs["stdin"] = subprocess.PIPE popen_kwargs["creationflags"] = ( popen_kwargs.get("creationflags", 0) | _CREATE_NEW_PROCESS_GROUP @@ -471,10 +461,7 @@ def start_agent(port: int = AGENT_SERVER_PORT) -> subprocess.Popen: _write_server_process_record(proc, port=port, server_py=server_py, server_host_source=host_source) def _stream_output() -> None: - # Size-capped copy (CPL4-C5): same bound as the server.log stdlib - # handler (2 MB live + numbered backups). Rotation failure must never - # kill the copy thread — worst case the live file keeps growing, which - # is exactly the pre-cap behavior. + # Match server.log rotation; rotation failure leaves streaming alive. max_bytes = 2 * 1024 * 1024 backups = 3 @@ -534,8 +521,20 @@ def stop_agent() -> None: log.info("Stopping agent (pid=%s)...", proc.pid) try: - # Graceful phase signals only the server: it owns its Manager and workers (#1142). - proc.terminate() + # Windows terminate is forced: its owned stdin asks the server to drain first. + if IS_WINDOWS and proc.stdin is not None: + try: + proc.stdin.write(b"quit\n") + proc.stdin.flush() + except (BrokenPipeError, OSError): + log.warning("Launcher stop pipe unavailable; retaining the exit wait and fallback") + finally: + try: + proc.stdin.close() + except OSError: + pass # A failed pipe must still reach the wait/forced cleanup. + else: + proc.terminate() proc.wait(timeout=LAUNCHER_STOP_GRACE_SEC) except subprocess.TimeoutExpired: if IS_WINDOWS and job is not None: diff --git a/ouroboros/gateway/control.py b/ouroboros/gateway/control.py index 96fab6023..6510de25e 100644 --- a/ouroboros/gateway/control.py +++ b/ouroboros/gateway/control.py @@ -77,6 +77,7 @@ def _managed_update_payload(*, fetch: bool, include_tags: bool) -> dict[str, Any "phase": str(tx.get("phase") or ""), "task_id": str(tx.get("task_id") or ""), "restart_required": bool(tx.get("restart_required")), + **({"local_work_recovery": True} if tx.get("stash_restore") else {}), } if tx else {"active": False} @@ -511,22 +512,24 @@ def _fence_failure(blockers: list[str], stash_note: str = "") -> JSONResponse: def _rollback_fenced_update(reason: str, error: str, **extra: Any) -> JSONResponse: - from supervisor.update_merge import mark_update_tx_gate_blocked, rollback_managed_update + from supervisor.update_merge import active_update_tx, mark_update_tx_gate_blocked, rollback_managed_update update_progress.advance("rolling_back") ok, message = rollback_managed_update(reason) if ok: _respawn_workers_after_failed_update() return JSONResponse( - {"error": error, "rolled_back": True, "rollback": message, **extra}, + {"error": error, "rolled_back": True, "rollback": message, "stash_note": message, **extra}, status_code=409, ) - mark_update_tx_gate_blocked(reason, message) + if not active_update_tx().get("stash_restore"): + mark_update_tx_gate_blocked(reason, message) return JSONResponse( { "error": error, "rolled_back": False, "rollback": message, + "stash_note": message, "restart_required": True, **extra, }, @@ -656,13 +659,13 @@ def _stash_local_work_fenced( def _unwind_stashed_update(tx: dict, context: str) -> str: """Undo the stash prologue when the update aborts before any repo mutation: - restore the exact stash entry (marker-guarded — a crash between the stash - apply and its drop must not let boot's replay wipe the already-restored - copy) and clear the tx. Returns a disclosure note ("" when clean).""" + restore the exact stash and clear only a confirmed restoration/preservation + outcome. An incomplete result retains its write-ahead marker for boot.""" from supervisor.update_merge import clear_update_tx, restore_stash_with_marker - note = restore_stash_with_marker(tx, context) - if not clear_update_tx(): + result = restore_stash_with_marker(tx, context) + note = result.note + if result.complete and not clear_update_tx(): note = (note + "; " if note else "") + "the update transaction marker could not be cleared" return note @@ -1101,7 +1104,18 @@ def _apply_smart_update_fenced( log.warning("managed smart update failed after writer fence", exc_info=True) from supervisor.update_merge import active_update_tx as _active_tx - if _active_tx(): + pending = _active_tx() + if pending.get("stash_restore"): + # A restore may already have returned files even if its completion + # record failed. Never route that uncertainty into generic rollback. + record = pending["stash_restore"] + return JSONResponse( + {"error": f"managed update failed: {type(exc).__name__}: {exc}", + "reason": "stash_recovery_incomplete", "restart_required": True, + "stash_note": record.get("note") or "Local work recovery is pending; current files were left unchanged."}, + status_code=500, + ) + if pending: return _rollback_fenced_update( "smart_update_exception", f"managed update failed: {type(exc).__name__}: {exc}", diff --git a/ouroboros/server_process.py b/ouroboros/server_process.py index 470063c43..0e8990fcd 100644 --- a/ouroboros/server_process.py +++ b/ouroboros/server_process.py @@ -14,6 +14,7 @@ import logging import os import pathlib import re +import sys import threading from contextlib import nullcontext from typing import Callable @@ -40,6 +41,7 @@ _supervisor_stop = threading.Event() # PROCESS is exiting. Unlike ``_supervisor_stop`` it is never cleared — a settings # save that lands mid-teardown must not revive a supervisor generation (#1142). _exit_signalled = threading.Event() +_stop_source = "external_signal" # Set only when the OWNER asked for the restart (the chat Restart button, and the @@ -342,6 +344,11 @@ def runtime_service_identity(drive_root: pathlib.Path, port: int, return _custodied_local_model(drive_root, port, host_matches) +def server_stop_source() -> str: + """Diagnostic stop ingress; no effect on custody or restart policy.""" + return _stop_source + + class _SignalStopServer(uvicorn.Server): """uvicorn.Server whose SIGTERM/SIGINT handler stops the supervisor loop AT THE SIGNAL. @@ -354,11 +361,45 @@ class _SignalStopServer(uvicorn.Server): the loop, and the lifespan ``finally`` still sets the event on every path. """ + def request_shutdown(self) -> None: + """The same irreversible stop for OS signals and the launcher's owned pipe.""" + global _stop_source + _stop_source = "launcher_quit" + _exit_signalled.set() + _supervisor_stop.set() + self.should_exit = True + def handle_exit(self, sig: int, frame) -> None: + # Let uvicorn see the previous should_exit value (repeated SIGINT forces exit). + global _stop_source + _stop_source = "external_signal" _exit_signalled.set() _supervisor_stop.set() super().handle_exit(sig, frame) + def watch_launcher_stop(self) -> None: + """Windows has no SIGTERM handler path; consume its launcher's private stdin. + + Opt-in only: direct servers retain their stdin. EOF is not a Quit command. + Consume the environment flag before spawning workers so their children cannot + mistake inherited stdin for a launcher command channel. Older launchers still + use their existing signals/forced-stop behavior until their package is updated. + """ + if os.environ.pop("OUROBOROS_LAUNCHER_STOP_STDIN", "") != "1": + return + stream = getattr(sys.stdin, "buffer", None) + if stream is None: + return + + def receive() -> None: + try: + if stream.readline(5) == b"quit\n": + self.request_shutdown() + except (OSError, ValueError): + log.warning("Launcher stop pipe closed without a Quit request", exc_info=True) + + threading.Thread(target=receive, name="launcher-stop", daemon=True).start() + def _embedded_uvicorn_server(config: "uvicorn.Config") -> "uvicorn.Server": """A uvicorn.Server hosted INSIDE the main server's event loop (the Host Service). diff --git a/ouroboros/server_restart.py b/ouroboros/server_restart.py index c7124a5d5..9943e280d 100644 --- a/ouroboros/server_restart.py +++ b/ouroboros/server_restart.py @@ -310,6 +310,12 @@ def _safe_restart_serialized(safe_restart_fn, *, reason: str, unsynced_policy: s return False, "Managed update state is unreadable; restart was deferred." if status == "future": return False, "Managed update state was recorded by a newer version; restart was deferred." + if status == "valid" and tx.get("stash_restore"): + return False, ( + "Local changes are still being recovered. Quit and reopen the desktop app, " + "or restart the server process for a web deployment. Restart was deferred " + "to preserve the current files." + ) if status == "absent" and not git_ops._clear_update_intent(): return False, ( "An update intent marker with no update transaction could not be removed; " diff --git a/server.py b/server.py index f9ffb57fb..2e5eee9f9 100644 --- a/server.py +++ b/server.py @@ -45,7 +45,7 @@ from ouroboros.server_process import ( # noqa: F401 _request_restart_exit, _restart_requested, _supervisor_stop, _exit_signalled, _SignalStopServer, _embedded_uvicorn_server, - capture_server_source_baseline, log, + capture_server_source_baseline, server_stop_source, log, ) from ouroboros.server_routing_context import ( # noqa: F401 _active_direct_roots, @@ -1497,7 +1497,7 @@ async def lifespan(app): { "ts": utc_now_iso(), "type": "server_shutdown", - "cause": "restart_requested" if restart_requested else "external_signal", + "cause": "restart_requested" if restart_requested else server_stop_source(), "restart_exit": restart_requested, }, ) @@ -1699,11 +1699,11 @@ def main() -> int: log_level="warning", ws_ping_interval=20, ws_ping_timeout=20, - # Bound the open HTTP/WS drain so the lifespan teardown (terminal custody) starts inside - # the launcher's stop budget instead of leaving terminalization to the next boot (#1142). + # Leave time for terminal custody inside the launcher stop budget (#1142). timeout_graceful_shutdown=SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_SEC, ) server = _SignalStopServer(config) + server.watch_launcher_stop() _uvicorn_exited = threading.Event() def _check_restart(): diff --git a/supervisor/update_candidate.py b/supervisor/update_candidate.py index b4c2e0c01..423ca9c25 100644 --- a/supervisor/update_candidate.py +++ b/supervisor/update_candidate.py @@ -13,7 +13,8 @@ Depends on ``git_ops`` via the module object (``_g.X``) so monkeypatched from __future__ import annotations import os -from typing import Any, Dict, List, Optional, Tuple +from dataclasses import dataclass, field +from typing import Any, Dict, List, Literal, Optional, Tuple from supervisor import git_ops as _g @@ -151,10 +152,13 @@ def live_unmerged_paths() -> Optional[List[str]]: """The live repo's unmerged (conflicted) paths; ``None`` when Git itself failed (an unreadable inventory must never masquerade as "no conflicts" — callers keep their previous list or fail safe).""" - rc_u, unmerged_out, _ue = _g.git_capture(["git", "diff", "--name-only", "--diff-filter=U"]) - if rc_u != 0: + rc_u, unmerged_out, _ue = _g._run_git_process_bounded( + ["git", "diff", "--name-only", "--diff-filter=U", "-z"], + timeout=_GIT_RUN_TIMEOUT_SEC, cwd=_g.REPO_DIR, text=False, + ) + if rc_u != 0 or not isinstance(unmerged_out, bytes): return None - return [ln.strip() for ln in unmerged_out.splitlines() if ln.strip()] + return [os.fsdecode(path) for path in unmerged_out.split(b"\0") if path] def _staged_blobs_batch(paths: List[bytes]) -> Tuple[Optional[Dict[bytes, bytes]], bytes]: @@ -540,118 +544,144 @@ def stash_local_changes_for_update(attempt_id: str) -> Tuple[str, str, str]: # later. The caller fail-closes if the tree still reports dirty. return "ok", "", "" + +@dataclass(frozen=True) +class StashRestoreResult: + """Effect, not a success-shaped note: only complete outcomes may clear a tx.""" + + status: Literal["not_needed", "restored", "preserved", "incomplete"] + note: str = "" + evidence: Dict[str, Any] = field(default_factory=dict) + + @property + def complete(self) -> bool: + return self.status != "incomplete" + + +def _stash_recovery_note(stash_sha: str) -> str: + return (f"the original local changes remain in stash {stash_sha[:12]} " + f"(rescue-local-{stash_sha[:12]}); inspect the current files before " + f"recovering with `git stash apply {stash_sha}`") + + def restore_update_stash( - stash_sha: str, context: str = "", on_applied=None -) -> Tuple[bool, str]: - """Apply-then-drop the exact update stash entry (matched by SHA). + stash_sha: str, context: str = "", *, before_cleanup=None, +) -> StashRestoreResult: + """Apply the exact pinned object; NEVER automatically drop an update stash. + + Transaction callers record intent before entering here and retain diagnostics + through ``before_cleanup`` before resetting a failed partial apply. Cleanup is + allowed only after a clean starting tree and is itself verified. A missing list + entry says nothing about restoration: the exact pinned object is the carrier. + """ + from supervisor.update_merge import create_rescue_local_ref - On an apply conflict the partial apply is reset away and the stash entry is - KEPT, so local work is never lost — the returned note tells the owner the - exact `git stash apply` command. Restoring onto the pre-update tree (the - rollback path) always applies cleanly because the stash was taken there. - apply+drop (NOT pop) keeps this crash-idempotent: dying between apply and - drop leaves the entry in place for the replay. ``on_applied`` runs between - the successful apply and the drop so the caller can persist a durable - "restored" marker that survives a crash in that window.""" if not stash_sha: - return True, "" - rc_l, listing, list_error = _g.git_capture(["git", "stash", "list", "--format=%H %gd"]) - if rc_l != 0: - return False, list_error or "could not list stash entries" - ref = "" - for line in listing.splitlines(): - sha, _sep, name = line.strip().partition(" ") - if sha == stash_sha and name: - ref = name - break - if not ref: - return True, ( - "stash entry not found in the stash list (already consumed after a " - "verified restore, or dropped externally)" - ) - # Apply by EXACT SHA (git stash apply accepts any stash-shaped commit): a - # concurrent stash push/drop can shift the stash@{n} selector between the - # list and the apply, and a selector-based apply would then touch someone - # else's entry. - rc_p, _po, apply_error = _g.git_capture(["git", "stash", "apply", stash_sha]) - if rc_p == 0: - if on_applied is not None: - try: - on_applied() - except Exception: - _g.log.warning("restore_update_stash on_applied hook failed", exc_info=True) - # Drop needs a selector, and selectors shift under concurrent stash - # traffic. Drop ONLY when the list is byte-identical to the pre-apply - # snapshot (nothing external happened) and the selector still names our - # exact SHA; otherwise keep the entry — an undropped own entry is - # harmless litter, a dropped foreign entry is someone's lost work. - rc_l2, listing2, _le2 = _g.git_capture(["git", "stash", "list", "--format=%H %gd"]) - drop_ref = "" - if rc_l2 == 0 and listing2 == listing: - for line in listing2.splitlines(): - sha2, _sep2, name2 = line.strip().partition(" ") - if sha2 == stash_sha and name2: - drop_ref = name2 - break - if drop_ref: - _g.git_capture(["git", "stash", "drop", drop_ref]) - else: - _g.log.info("stash entry %s kept (list changed during restore)", stash_sha[:12]) - from supervisor.update_merge import _log_supervisor + return StashRestoreResult("not_needed") + if _rev_parse(stash_sha) != stash_sha or not create_rescue_local_ref(stash_sha): + return StashRestoreResult("incomplete", f"could not verify and pin update stash {stash_sha}") + recovery = _stash_recovery_note(stash_sha) + rc_s, dirty, status_error = _g.git_capture(["git", "status", "--porcelain"]) + if rc_s != 0: + return StashRestoreResult("incomplete", f"could not inspect current files: {status_error}; {recovery}") + if dirty: + return StashRestoreResult("preserved", f"local changes are present, so the stash was NOT auto-applied; {recovery}") + rc, stdout, stderr = _g.git_capture(["git", "stash", "apply", stash_sha]) + evidence = {"apply_returncode": rc, "stdout": stdout, "stderr": stderr} + if rc == 0: + return StashRestoreResult("restored", "local changes restored; the update stash was retained", evidence) + conflicts = live_unmerged_paths() + evidence["conflict_paths"] = conflicts + if conflicts: + import json - _log_supervisor({ - "type": "managed_update_stash_restored", - "context": context, - "stash_sha": stash_sha, - }) - return True, "local changes restored" - _g.git_capture(["git", "reset", "--hard", "HEAD"]) - _g.git_capture(["git", "clean", "-fd"]) - note = ( - "local changes could not be restored automatically " - f"({(apply_error or '').strip() or 'conflict with the updated tree'}); they are " - f"preserved in git stash entry {stash_sha[:12]} — recover with " - f"`git stash apply {stash_sha}`" - ) - from supervisor.update_merge import _log_supervisor + problem = "conflicting paths: " + ", ".join(json.dumps(path, ensure_ascii=False) for path in conflicts) + elif conflicts is None: + problem = "git stash apply failed; the conflict inventory could not be read" + else: + problem = "git stash apply failed without tracked conflicts (untracked collisions are not in that inventory)" + diagnostics = "\n".join(part for part in (stdout, stderr) if part) + note = f"{problem}; {recovery}" + (f"\nGit diagnostics:\n{diagnostics}" if diagnostics else "") + if before_cleanup is not None: + before_cleanup({"status": "cleanup_pending", "note": note, **evidence}) + cleanup = {} + for name, cmd in (("reset", ["git", "reset", "--hard", "HEAD"]), + ("clean", ["git", "clean", "-fd"])): + code, out, err = _g.git_capture(cmd) + cleanup[name] = {"returncode": code, "stdout": out, "stderr": err} + if code != 0: + return StashRestoreResult("incomplete", f"{note}\nPartial-apply cleanup failed at {name}: {err or out}", + {**evidence, "cleanup": cleanup}) + rc_s, dirty, status_error = _g.git_capture(["git", "status", "--porcelain"]) + evidence["cleanup"] = {**cleanup, "verified_clean": rc_s == 0 and not dirty} + if rc_s != 0 or dirty: + return StashRestoreResult("incomplete", f"{note}\nPartial-apply cleanup was not verified: {status_error or dirty}", evidence) + return StashRestoreResult("preserved", note, evidence) - _log_supervisor({ - "type": "managed_update_stash_restore_failed", - "context": context, - "stash_sha": stash_sha, - "error": (apply_error or "").strip(), - }) - return False, note -def restore_stash_with_marker(tx: Dict[str, Any], context: str) -> str: - """Marker-guarded stash restore for every tx-clearing path: skips when a prior - attempt already restored (a crash between apply and drop must not let a REPLAY - conflict against the already-restored copy and reset it away), and persists the - ``stash_restored`` marker between apply and drop. Returns the disclosure note.""" +def restore_stash_with_marker(tx: Dict[str, Any], context: str) -> StashRestoreResult: + """One recovery owner for all completion paths, with write-ahead custody. + + ``stash_restore`` is schema-2 evidence: old strict readers refuse it. Legacy + ``stash_restored`` still means a successful recorded apply, never its intent. + An interrupted apply/cleanup is NOT replayed: once the store works again the + pinned backup is handed back with an honest unknown-outcome note, preserving + both the current files and any subsequent owner edits. + """ + from supervisor import update_merge as _um + stash_sha = str(tx.get("stash_sha") or "") - if not stash_sha or bool(tx.get("stash_restored")): - return "" - # Restoring onto a DIRTY tree is never safe: a conflicting apply's cleanup - # (reset --hard + clean -fd) would wipe whatever made the tree dirty — late - # human edits in an abort-unwind, an operator's work on a diverged head. - # Rollback/finalize call this on verified-clean trees, so this guard only - # fires where the destructive cleanup would actually cost something. - rc_s, dirty, _se = _g.git_capture(["git", "status", "--porcelain"]) - if rc_s != 0 or dirty.strip(): - return ( - "local changes are present, so the stashed work was NOT auto-applied; it is " - f"preserved in git stash entry {stash_sha[:12]} — recover with " - f"`git stash apply {stash_sha}`" + record = tx.get("stash_restore") or {} + if not stash_sha: + return StashRestoreResult("not_needed") + if not record and tx.get("stash_restored"): + return StashRestoreResult("restored") # existing schema-1 success evidence + + def persist(value: Dict[str, Any]) -> None: + updated = {**tx, "stash_restore": {"context": context, "stash_sha": stash_sha, **value}} + if value["status"] == "restored": + updated["stash_restored"] = True + _um.write_update_tx(updated) + tx.update(updated) # never publish an in-memory success before the write + + try: + if record.get("status") in {"restored", "preserved"}: + result = StashRestoreResult(record["status"], record.get("note", ""), record.get("evidence", {})) + elif record: + if _rev_parse(stash_sha) != stash_sha or not _um.create_rescue_local_ref(stash_sha): + return StashRestoreResult("incomplete", f"interrupted restore: could not verify and pin stash {stash_sha}") + result = StashRestoreResult( + "preserved", "the automatic restore was interrupted; its result is unconfirmed and current files were left unchanged; " + + _stash_recovery_note(stash_sha) + + ("\n" + record["note"] if record.get("note") else ""), + {"interrupted": record}, + ) + persist({"status": result.status, "note": result.note, "evidence": result.evidence}) + else: + persist({"status": "applying"}) + result = restore_update_stash(stash_sha, context, before_cleanup=persist) + persist({"status": result.status, "note": result.note, "evidence": result.evidence}) + except Exception as exc: + _g.log.warning("update stash recovery could not be recorded", exc_info=True) + return StashRestoreResult( + "incomplete", f"automatic stash recovery remains unconfirmed ({type(exc).__name__}: {exc}); " + "current files must not be reset or re-applied; " + _stash_recovery_note(stash_sha), ) - - def _mark() -> None: - from supervisor import update_merge as _um - - tx["stash_restored"] = True - _um.write_update_tx(tx) - - _restored, note = restore_update_stash(stash_sha, context=context, on_applied=_mark) - return note + if not result.complete: + return result + event = {"type": "managed_update_stash_restored" if result.status == "restored" else "managed_update_stash_restore_failed", + "context": context, "stash_sha": stash_sha, "restore_status": result.status, + "stash_note": result.note, **result.evidence} + # The existing event is the durable disclosure after the marker is removed. + # A failed append retains the terminal marker; retrying it never repeats Git. + try: + recorded = _um._log_supervisor(event) + except Exception: + recorded = False + _g.log.warning("update stash recovery event failed", exc_info=True) + if not recorded: + return StashRestoreResult("incomplete", result.note + "; recovery event could not be saved", result.evidence) + return result def destructive_apply_guard(branch: str, pre_update_sha: str) -> str: diff --git a/supervisor/update_merge.py b/supervisor/update_merge.py index fda9c494e..bfed7f026 100644 --- a/supervisor/update_merge.py +++ b/supervisor/update_merge.py @@ -32,7 +32,7 @@ from supervisor.update_candidate import ( # noqa: F401 find_update_stash_sha, restore_stash_with_marker, restore_update_stash, stash_local_changes_for_update, lookup_update_stash, destructive_apply_guard, project_version_carriers, - quarantine_corrupt_update_tx_marker, + quarantine_corrupt_update_tx_marker, StashRestoreResult, ) # The planner, the clean-plan commit builder and the live materializer — the # carrier engine's three insertion points — live in supervisor.update_merge_plan @@ -44,16 +44,13 @@ from supervisor.update_merge_plan import ( # noqa: F401 UPDATE_TX_MARKER_NAME = "ouroboros-update-tx.json" -# ABI 7.0 / F14 N−1 transition shim: every tx write stamps the marker with the -# shared ``_schema_version`` key. ``finalize_managed_update_on_boot`` is the -# stable N−1→N entry point — an UNSTAMPED marker is the accepted pre-7.0 form -# (a version-N updater must finalize the transaction its N−1 predecessor -# recorded), a stamp ≤ ours is current, and an integer stamp ABOVE ours means -# a NEWER release recorded the transaction: this version cannot interpret it, -# so every strict reader fails closed (``"future"`` — never rolled back, left -# for the owner). A rolled-back N−1 binary reads a stamped marker unchanged: -# the stamp is one additive key its reader ignores. -UPDATE_TX_SCHEMA_VERSION = 1 +# Strict readers accept legacy unstamped/schema-1 transactions. Only the new +# stash-restore write-ahead state needs schema 2: a rollback can check out OLD +# code before restoring work, and a schema-1 reader must refuse that state, +# never ignore it and reset already-returned files. Ordinary pre-restore crash +# recovery remains schema 1. Backward refusal is not automatic recovery; an +# interrupted schema-2 restore needs this reader or deliberate owner recovery. +UPDATE_TX_SCHEMA_VERSION = 2 # ``dict.get`` sentinel distinguishing a MISSING ``_schema_version`` key (the # accepted pre-7.0 unstamped form) from an explicit stored ``null`` (corrupt). @@ -90,7 +87,7 @@ def write_update_tx(payload: Dict[str, Any]) -> None: atomic_write_json( _update_tx_marker_path(), - with_schema_version(payload, UPDATE_TX_SCHEMA_VERSION), + with_schema_version(payload, 2 if payload.get("stash_restore") else 1), trailing_newline=True, ) @@ -108,8 +105,8 @@ def clear_update_tx() -> bool: _ASSISTED_PHASES = ("materializing_assisted", "assisted_resolution", "committing_assisted") GATE_BLOCKED_PHASE = "gate_blocked" -# The repository is ALREADY in its final good state and only the tx-marker -# unlink failed: recovery retries the cleanup and must never roll back. A +# The completion outcome is recorded and only the tx-marker unlink failed: +# recovery retries that cleanup and must never roll back. A # separate phase — not a gate_blocked sub-kind — because gate_blocked's boot # contract ("a refused merge: fresh rollback, never promotion") stays exact. MARKER_CLEANUP_RETRY_PHASE = "marker_cleanup_retry" @@ -440,12 +437,22 @@ def rollback_managed_update( branch = str(tx.get("pre_update_branch") or _g.BRANCH_DEV) if not pre: return False, "no pre_update_sha in update tx marker" + # Read the write-ahead fact BEFORE phase writes or any reset. A restore + # may have changed files even when its final marker never reached disk. + if tx.get("stash_restore"): + result = restore_stash_with_marker(tx, "rollback_resume") + rc_h, head, _he = _g.git_capture(["git", "rev-parse", "--verify", "HEAD"]) + rc_b, current_branch, _be = _g.git_capture(["git", "rev-parse", "--abbrev-ref", "HEAD"]) + if rc_h != 0 or head != pre or rc_b != 0 or current_branch != branch: + return False, "rollback left the current checkout unchanged after stash recovery; " + result.note + if not _g._clear_update_intent(): + return False, "rollback could not clear update intent; " + result.note + return _finish_rollback(tx, pre, branch, reason, result, reopen_writer_admission) tx["phase"] = "rolling_back" tx["rollback_reason"] = str(reason or "update_rollback") write_update_tx(tx) - # Q1=C replay guard: a previous rollback attempt that already RESTORED the - # owner's stashed work (durable stash_restored marker, written between the - # stash apply and its drop) must not reset that work away on resume. When + # Legacy replay guard: a durable stash_restored marker proves a completed + # apply. It retains that meaning for older transactions. When # HEAD already sits at the pre-update SHA on the right branch, the repo part # of the rollback is done — skip the destructive re-reset and re-restore. resume_with_restored_work = False @@ -460,7 +467,7 @@ def rollback_managed_update( if not _g._clear_update_intent(): return False, "rollback could not clear update intent before restoring the repository" if resume_with_restored_work: - return _finish_rollback(tx, pre, branch, reason, "", reopen_writer_admission) + return _finish_rollback(tx, pre, branch, reason, StashRestoreResult("restored"), reopen_writer_admission) # Fresh rescue BEFORE the destructive reset; the persisted pointer is the replay guard. if not tx.get("rollback_rescue"): _g.rescue_into_tx(tx, key="rollback_rescue", reason=str(reason), @@ -511,31 +518,32 @@ def rollback_managed_update( ): detail = head_error or branch_error or status_error or "rollback verification mismatch" return _fail(f"rollback could not be verified: {detail}") - # Q1=C: the pre-update tree is exactly where the stash was taken, so this - # restore is conflict-free; a kept stash entry is disclosed, never dropped. - stash_note = restore_stash_with_marker(tx, "rollback") - return _finish_rollback(tx, pre, branch, reason, stash_note, reopen_writer_admission) + result = restore_stash_with_marker(tx, "rollback") + return _finish_rollback(tx, pre, branch, reason, result, reopen_writer_admission) def _finish_rollback( tx: Dict[str, Any], pre: str, branch: str, reason: str, - stash_note: str, reopen_writer_admission: bool, + result: StashRestoreResult, reopen_writer_admission: bool, ) -> Tuple[bool, str]: """Shared rollback tail: close admission, clear the tx, log, reopen, report.""" + if not result.complete: + return False, result.note + stash_note = result.note gate_reason = "managed_update:rollback" try: from supervisor.workers import close_repo_writer_admission close_repo_writer_admission(gate_reason) except Exception: - return False, "rollback restored the repository but could not close writer admission" + return False, "rollback restored the repository but could not close writer admission; " + stash_note if not clear_update_tx(): - return False, "rollback restored the repository but could not clear update transaction" + return False, "rollback restored the repository but could not clear update transaction; " + stash_note rescue = tx.get("rollback_rescue") if isinstance(tx.get("rollback_rescue"), dict) else {} - append_jsonl( - _g.DRIVE_ROOT / "logs" / "supervisor.jsonl", - {"ts": utc_now_iso(), "type": "managed_update_rolled_back", "reason": reason, + _log_supervisor( + {"type": "managed_update_rolled_back", "reason": reason, "pre_update_sha": pre, "branch": branch, + "stash_note": stash_note, "stash_restore_status": result.status, **({"failed_update_ref": tx["failed_update_ref"]} if tx.get("failed_update_ref") else {}), **{f"rescue_{key}": rescue[key] for key in ("path", "ref", "ts") if rescue.get(key)}, **({"rescue_error": tx["rollback_rescue_error"]} if tx.get("rollback_rescue_error") else {})}, @@ -805,8 +813,28 @@ def external_host_restart_smoke() -> Dict[str, Any]: return _run_update_smoke([sys.executable, hook, "--check"]) -def _log_supervisor(payload: Dict[str, Any]) -> None: - append_jsonl(_g.DRIVE_ROOT / "logs" / "supervisor.jsonl", {"ts": utc_now_iso(), **payload}) +def _log_supervisor(payload: Dict[str, Any]) -> bool: + try: + return append_jsonl(_g.DRIVE_ROOT / "logs" / "supervisor.jsonl", {"ts": utc_now_iso(), **payload}) + except Exception: + _g.log.warning("managed update event could not be saved", exc_info=True) + return False + + +def _complete_stash_recovery( + tx: Dict[str, Any], context: str, result: Dict[str, Any], event: Dict[str, Any], +) -> Dict[str, Any]: + """The three boot completion paths share effect-checked transaction cleanup.""" + restored = restore_stash_with_marker(tx, context) + disclosure = ({"stash_note": restored.note, "stash_restore_status": restored.status} + if restored.note else {}) + if not restored.complete: + return {"finalized": False, "reason": "stash_recovery_incomplete", **disclosure} + if not clear_update_tx(): + mark_update_tx_gate_blocked("stash_marker_cleanup_failed", cleanup_only=True) + return {"finalized": False, "reason": "could not clear update marker", **disclosure} + _log_supervisor({**event, **disclosure}) + return {**result, **disclosure} def _finalize_pending_boot_smoke(tx: Dict[str, Any], supervisor_ready: bool) -> Dict[str, Any]: @@ -837,7 +865,7 @@ def _finalize_pending_boot_smoke(tx: Dict[str, Any], supervisor_ready: bool) -> "msg": msg, "smoke": smoke, }) - return {"finalized": False, "rolled_back": ok, "msg": msg, "smoke": smoke} + return {"finalized": False, "rolled_back": ok, "msg": msg, "stash_note": msg, "smoke": smoke} tx["pre_restart_smoke"] = _PRE_RESTART_SMOKE_PASSED write_update_tx(tx) if bool(supervisor_ready) and merge_in_head: @@ -848,37 +876,20 @@ def _finalize_pending_boot_smoke(tx: Dict[str, Any], supervisor_ready: bool) -> ) _log_supervisor({"type": "managed_update_native_host_failed", "ok": ok, "msg": msg, "smoke": native_smoke}) - return {"finalized": False, "rolled_back": ok, "msg": msg, "smoke": native_smoke} + return {"finalized": False, "rolled_back": ok, "msg": msg, "stash_note": msg, "smoke": native_smoke} if not _g._clear_update_intent(): mark_update_tx_gate_blocked("finalize_intent_cleanup_failed") return {"finalized": False, "reason": "could not clear update intent"} - # Q1=C: the boot survived on the merged tree — bring the owner's stashed - # local work back as uncommitted content BEFORE clearing the tx, so the - # durable stash_sha pointer survives a crash in this window (a replayed - # finalize sees "already consumed" and the restored tree stays). A - # conflicting restore keeps the stash entry and the note discloses the - # manual recovery command. - stash_note = restore_stash_with_marker(tx, "boot_finalize") - if not clear_update_tx(): - mark_update_tx_gate_blocked("finalize_marker_cleanup_failed", cleanup_only=True) - return {"finalized": False, "reason": "could not clear update marker", - **({"stash_note": stash_note} if stash_note else {})} - _log_supervisor({ - "type": "managed_update_finalized", - "head": head, - **({"stash_note": stash_note} if stash_note else {}), - }) - result = {"finalized": True} - if stash_note: - result["stash_note"] = stash_note - return result + return _complete_stash_recovery( + tx, "boot_finalize", {"finalized": True}, {"type": "managed_update_finalized", "head": head}, + ) if (bool(supervisor_ready) and head_resolved and not merge_in_head) or attempts >= 2: ok, msg = rollback_managed_update( "post_boot_smoke_failed", reopen_writer_admission=False ) _log_supervisor({"type": "managed_update_rollback_after_failed_boot", "ok": ok, "msg": msg, "boot_attempts": attempts}) - return {"finalized": False, "rolled_back": ok, "msg": msg} + return {"finalized": False, "rolled_back": ok, "msg": msg, "stash_note": msg} tx["boot_attempts"] = attempts write_update_tx(tx) return {"finalized": False, "boot_attempts": attempts} @@ -899,7 +910,7 @@ def _recover_assisted_on_boot(tx: Dict[str, Any], supervisor_ready: bool) -> Dic ) _log_supervisor({"type": "managed_update_assisted_materialization_interrupted", "ok": ok, "msg": msg}) - return {"finalized": False, "rolled_back": ok, "msg": msg} + return {"finalized": False, "rolled_back": ok, "msg": msg, "stash_note": msg} if state == "committed": # Only pending_boot_smoke proves that exact-binding verification, post-commit # tests, and the pre-restart smoke all passed. A commit found under an assisted @@ -909,21 +920,17 @@ def _recover_assisted_on_boot(tx: Dict[str, Any], supervisor_ready: bool) -> Dic ) _log_supervisor({"type": "managed_update_assisted_commit_unproven", "ok": ok, "msg": msg}) - return {"finalized": False, "rolled_back": ok, "msg": msg} + return {"finalized": False, "rolled_back": ok, "msg": msg, "stash_note": msg} if state == "diverged": # A real reviewed commit landed; keep it (never reset over reviewed work), abandon # this update — it is re-planned fresh later. The tx is the ONLY pointer that a # stash restore is owed: bring the owner's stashed work back (a conflicting # restore keeps the entry and the note discloses it) BEFORE dropping the marker. - stash_note = restore_stash_with_marker(tx, "boot_diverged_abandon") - if not clear_update_tx(): - mark_update_tx_gate_blocked("diverged_marker_cleanup_failed", cleanup_only=True) - return {"finalized": False, "reason": "could not clear diverged update marker", - **({"stash_note": stash_note} if stash_note else {})} - _log_supervisor({"type": "managed_update_assisted_abandoned_diverged", - **({"stash_note": stash_note} if stash_note else {})}) - return {"finalized": False, "abandoned": True, "reason": "head_diverged", - **({"stash_note": stash_note} if stash_note else {})} + return _complete_stash_recovery( + tx, "boot_diverged_abandon", + {"finalized": False, "abandoned": True, "reason": "head_diverged"}, + {"type": "managed_update_assisted_abandoned_diverged"}, + ) if state == "in_progress": attempts = int(tx.get("resolution_attempts") or 0) + 1 if attempts > _ASSISTED_BOOT_ATTEMPT_CAP: @@ -931,7 +938,7 @@ def _recover_assisted_on_boot(tx: Dict[str, Any], supervisor_ready: bool) -> Dic "assisted_resolution_expired", reopen_writer_admission=False ) _log_supervisor({"type": "managed_update_assisted_expired", "ok": ok, "msg": msg}) - return {"finalized": False, "rolled_back": ok, "msg": msg} + return {"finalized": False, "rolled_back": ok, "msg": msg, "stash_note": msg} # Re-establish the merge state if the restart/rescue wiped it; preserve partial # progress when MERGE_HEAD + a dirty tree already survived. rc_d, dirty, _de = _g.git_capture(["git", "status", "--porcelain"]) @@ -954,7 +961,7 @@ def _recover_assisted_on_boot(tx: Dict[str, Any], supervisor_ready: bool) -> Dic ) _log_supervisor({"type": "managed_update_backfill_projection_failed", "error": bp_error, "rollback": rb_msg}) - return {"finalized": False, "rolled_back": rb_ok, "msg": bp_error} + return {"finalized": False, "rolled_back": rb_ok, "msg": bp_error, "stash_note": rb_msg} m0_backfill, m0_err = worktree_snapshot_tree("HEAD") if m0_backfill: tx["m0_tree"] = m0_backfill @@ -1000,7 +1007,7 @@ def _recover_assisted_on_boot(tx: Dict[str, Any], supervisor_ready: bool) -> Dic ) _log_supervisor({"type": "managed_update_assisted_rematerialize_failed", "materialize_msg": msg, "rollback": rb_msg}) - return {"finalized": False, "rolled_back": rb_ok, "msg": msg} + return {"finalized": False, "rolled_back": rb_ok, "msg": msg, "stash_note": rb_msg} tx["phase"] = "assisted_resolution" tx["resolution_attempts"] = attempts write_update_tx(tx) @@ -1199,7 +1206,7 @@ def abort_orphaned_assisted_tx( ) except Exception: _g.log.warning("abort_orphaned_assisted_tx: worker pool start failed", exc_info=True) - return {"acted": True, "rolled_back": ok, "msg": msg} + return {"acted": True, "rolled_back": ok, "msg": msg, "stash_note": msg} finally: if lock_fh is not None: release_update_lock(lock_fh) @@ -1215,7 +1222,7 @@ def _recover_replace_on_boot(tx: Dict[str, Any], supervisor_ready: bool) -> Dict ok, msg = rollback_managed_update( "replace_apply_state_unreadable", reopen_writer_admission=False ) - return {"finalized": False, "rolled_back": ok, "msg": msg} + return {"finalized": False, "rolled_back": ok, "msg": msg, "stash_note": msg} applied = head == target and not dirty.strip() if applied: tx["phase"] = "pending_boot_smoke" @@ -1236,7 +1243,7 @@ def _recover_replace_on_boot(tx: Dict[str, Any], supervisor_ready: bool) -> Dict ok, msg = rollback_managed_update( "replace_apply_interrupted", reopen_writer_admission=False ) - return {"finalized": False, "rolled_back": ok, "msg": msg} + return {"finalized": False, "rolled_back": ok, "msg": msg, "stash_note": msg} def finalize_managed_update_on_boot(supervisor_ready: bool = True) -> Dict[str, Any]: @@ -1280,6 +1287,17 @@ def finalize_managed_update_on_boot(supervisor_ready: bool = True) -> Dict[str, return {"finalized": False, "reason": "update tx recorded by a newer version — left for owner"} phase = str(tx.get("phase") or "") + if tx.get("stash_restore") and phase != "rolling_back": + # Restoration is the final repository operation in these phases. + # Do not rerun smoke/merge recovery over already-restored or later + # owner work. A usable store can finish the preservation handoff. + return _complete_stash_recovery( + tx, "boot_restore_resume", + {"finalized": (tx.get("gate_blocked_from_phase") if phase == MARKER_CLEANUP_RETRY_PHASE + else phase) == "pending_boot_smoke", + "reason": "resumed local work recovery"}, + {"type": "managed_update_stash_recovered_on_boot"}, + ) if phase == "stashing_local_work": # Crash between the durable pre-stash marker and the merge apply: # nothing was applied yet. Restore whatever the attempt stashed @@ -1294,19 +1312,13 @@ def finalize_managed_update_on_boot(supervisor_ready: bool = True) -> Dict[str, _log_supervisor({"type": "managed_update_stash_recovery_unreadable"}) return {"finalized": False, "reason": "stash storage unreadable — recovery left for a later boot"} - stash_note = "" if sha: tx["stash_sha"] = sha - stash_note = restore_stash_with_marker(tx, "boot_stash_recovery") - if not clear_update_tx(): - mark_update_tx_gate_blocked("stash_recovery_marker_cleanup_failed", cleanup_only=True) - return {"finalized": False, "reason": "could not clear update marker"} - _log_supervisor({ - "type": "managed_update_stash_recovered_on_boot", - **({"stash_note": stash_note} if stash_note else {}), - }) - return {"finalized": False, "reason": "recovered pre-apply stash crash", - **({"stash_note": stash_note} if stash_note else {})} + return _complete_stash_recovery( + tx, "boot_stash_recovery", + {"finalized": False, "reason": "recovered pre-apply stash crash"}, + {"type": "managed_update_stash_recovered_on_boot"}, + ) if phase == "pending_boot_smoke": return _finalize_pending_boot_smoke(tx, supervisor_ready) if phase == "applying_replace": @@ -1316,7 +1328,7 @@ def finalize_managed_update_on_boot(supervisor_ready: bool = True) -> Dict[str, str(tx.get("rollback_reason") or "boot_rollback_resume"), reopen_writer_admission=False, ) - return {"finalized": False, "rolled_back": ok, "msg": msg} + return {"finalized": False, "rolled_back": ok, "msg": msg, "stash_note": msg} if phase in _ASSISTED_PHASES: return _recover_assisted_on_boot(tx, supervisor_ready) if phase == MARKER_CLEANUP_RETRY_PHASE: @@ -1332,7 +1344,7 @@ def finalize_managed_update_on_boot(supervisor_ready: bool = True) -> Dict[str, str(tx.get("gate_blocked_reason") or "boot_gate_recovery"), reopen_writer_admission=False, ) - return {"finalized": False, "rolled_back": ok, "msg": msg} + return {"finalized": False, "rolled_back": ok, "msg": msg, "stash_note": msg} return {"finalized": False, "reason": f"unhandled phase {phase}"} except Exception: _g.log.warning("finalize_managed_update_on_boot failed", exc_info=True) diff --git a/supervisor/worker_pool_lifecycle.py b/supervisor/worker_pool_lifecycle.py index 4d8404e0e..7599e40b6 100644 --- a/supervisor/worker_pool_lifecycle.py +++ b/supervisor/worker_pool_lifecycle.py @@ -2,7 +2,8 @@ A spawned or respawned slot is installed unassignable (``reaping=True``) and opens only when the child's own ``worker_ready`` row is observed, which is also where the -SHA it booted is verified; a child alive but silent past the readiness window is +checkout SHA sampled after initialization is compared with the managed baseline; +a child alive but silent past the readiness window is torn down and replaced through the same respawn path, a bounded number of times. Its own entry-progress row permits one extension, still bounded from its birth. The pids workers ran under are recorded durably so an orphan surviving a restart @@ -20,6 +21,7 @@ import logging import json import os import pathlib +import subprocess import sys import threading import time @@ -51,6 +53,7 @@ def _pool(): _WORKER_LIFECYCLE_LOCK = threading.RLock() +_WORKER_SHA_LOOKUP_TIMEOUT_SEC = 2.0 # Local diagnostic I/O, like utils.get_git_info. def _serialized_worker_lifecycle(fn): @@ -265,8 +268,9 @@ def _verify_worker_sha_after_spawn( ``respawn_worker`` install a slot with ``reaping=True`` (the marker the assignment path, the crash detector and the reaper already honour) and hand it here. A slot opens only when the child's own ``worker_ready`` row - (supervisor/worker_process.py) names its pid, and that row's ``git_sha`` is - verified against ``current_sha`` in the same step. A child that is alive + (supervisor/worker_process.py) names its pid. That row samples the checkout + after initialization; comparison with ``current_sha`` is diagnostic only, + not loaded-code attestation, authorship or review evidence. A child that is alive but silent past ``WORKER_READY_WINDOW_SEC`` is torn down and replaced; its own ``worker_starting`` row permits one extension to ``WORKER_READY_CEILING_SEC`` from birth. Replacement still runs @@ -365,11 +369,48 @@ def _watch_booting_slots( pending.pop(wid) +def _worker_sha_relation(expected_sha: str, observed_sha: str) -> Dict[str, Any]: + """Read ancestry without fetching or the mutating git index-repair path. + + Only a negative answer needs a second read: shallow history cannot prove + non-ancestry, even when both commit objects are present locally. + """ + if not expected_sha: + return {"relation": "not_applicable"} + if not observed_sha: + return {"relation": "unavailable", "relation_error": "missing_observed_sha"} + if observed_sha == expected_sha: + return {"relation": "equal"} + options = dict(cwd=_pool().REPO_DIR, capture_output=True, text=True, + timeout=_WORKER_SHA_LOOKUP_TIMEOUT_SEC, + env={**os.environ, "GIT_NO_LAZY_FETCH": "1"}) + try: + result = subprocess.run( + ["git", "merge-base", "--is-ancestor", expected_sha, observed_sha], + **options, + ) + if result.returncode == 0: + return {"relation": "descendant"} + if result.returncode == 1: + shallow = subprocess.run(["git", "rev-parse", "--is-shallow-repository"], **options) + if shallow.returncode == 0 and shallow.stdout.strip() == "false": + # An older ancestor and a different branch both belong here. + return {"relation": "non_descendant"} + return {"relation": "unavailable", "relation_returncode": shallow.returncode, + "relation_error": ("shallow_history" if shallow.stdout.strip() == "true" + else "history_completeness_unavailable")} + except Exception as exc: + return {"relation": "unavailable", "relation_error": str(exc), + "relation_error_type": type(exc).__name__} + return {"relation": "unavailable", "relation_returncode": result.returncode, + "relation_error": result.stderr.strip()} + + def _open_ready_slot( wid: int, slot: Any, row: Dict[str, Any], expected_sha: str, owner_chat_id: int, started: float, attempt: int, ) -> None: - """The child confirmed ready: open the slot (if it is still ours) and verify its SHA.""" + """Open the ready slot before the diagnostic checkout comparison.""" with _queue_lock: owned = (_pool().WORKERS.get(wid) is slot and not getattr(slot, "readiness_exhausted", False)) @@ -377,6 +418,8 @@ def _open_ready_slot( slot.reaping = False observed_sha = str(row.get("git_sha") or "").strip() ok = (bool(observed_sha) and observed_sha == expected_sha) if expected_sha else None + wait_sec = round(time.time() - started, 2) + relation = _worker_sha_relation(expected_sha, observed_sha) _supervisor_row({ "type": "worker_sha_verify", "ok": ok, @@ -384,14 +427,19 @@ def _open_ready_slot( "observed_sha": observed_sha, "worker_pid": row.get("pid"), "worker_id": wid, - "wait_sec": round(time.time() - started, 2), + "wait_sec": wait_sec, "attempt": attempt, "slot_opened": owned, + **relation, }) - if ok is False and owner_chat_id: + if owner_chat_id and relation["relation"] in {"non_descendant", "unavailable"}: + observation = f"baseline {expected_sha[:8]}, observed {(observed_sha or 'unknown')[:8]}" + detail = ("is not a descendant of the recorded baseline" + if relation["relation"] == "non_descendant" + else "could not be compared with the recorded baseline") _pool().send_with_budget( owner_chat_id, - f"⚠️ Worker SHA mismatch after spawn: expected {expected_sha[:8]}, got {(observed_sha or 'unknown')[:8]}", + f"⚠️ Worker checkout sampled after initialization {detail}: {observation}.", role="system", system_type="worker_readiness_notice") diff --git a/tests/test_server_shutdown.py b/tests/test_server_shutdown.py index 18637a54d..15cbbb8d7 100644 --- a/tests/test_server_shutdown.py +++ b/tests/test_server_shutdown.py @@ -429,8 +429,13 @@ def test_main_normal_exit_does_not_run_emergency_cleanup(monkeypatch, tmp_path): class FakeServer: def __init__(self, _config): self.should_exit = False + self.stop_watcher_bound = False + + def watch_launcher_stop(self): + self.stop_watcher_bound = True def run(self, *, sockets): + assert self.stop_watcher_bound assert len(sockets) == 1 and sockets[0].getsockname()[1] > 0 return None @@ -463,8 +468,13 @@ def test_main_graceful_restart_cleanup_avoids_port_sweep(monkeypatch, tmp_path): class FakeServer: def __init__(self, _config): self.should_exit = False + self.stop_watcher_bound = False + + def watch_launcher_stop(self): + self.stop_watcher_bound = True def run(self, *, sockets): + assert self.stop_watcher_bound assert len(sockets) == 1 and sockets[0].getsockname()[1] > 0 server._restart_requested.set() return None diff --git a/tests/test_shutdown_signal_1142.py b/tests/test_shutdown_signal_1142.py index 3d609e813..70291ec96 100644 --- a/tests/test_shutdown_signal_1142.py +++ b/tests/test_shutdown_signal_1142.py @@ -168,6 +168,15 @@ def _wait_json(url: str, key: str, timeout_sec: float) -> None: @pytest.mark.serial @pytest.mark.skipif(sys.platform == "win32", reason="process groups and SIGTERM are POSIX") def test_group_sigterm_reaches_terminal_custody_without_a_false_supervisor_alarm(tmp_path): + _check_real_shutdown(tmp_path, launcher_pipe=False) + + +@pytest.mark.serial +def test_launcher_pipe_reaches_server_shutdown_without_a_signal(tmp_path): + _check_real_shutdown(tmp_path, launcher_pipe=True) + + +def _check_real_shutdown(tmp_path, *, launcher_pipe): """A real server in an isolated data root, SIGTERMed the way the OLD launcher does (whole group, so its Manager dies first) while a browser WebSocket is open: the lifespan teardown must still run inside the launcher budget and the owner chat must get no supervisor_failure.""" @@ -188,7 +197,8 @@ def test_group_sigterm_reaches_terminal_custody_without_a_false_supervisor_alarm "OUROBOROS_RUNTIME_MODE": "light", }), encoding="utf-8") # An owner chat is bound, so the false alarm WOULD be written if the crash counter fired. - (data_dir / "state" / "state.json").write_text(json.dumps({"owner_chat_id": 1}), encoding="utf-8") + from devtools.benchmarks.common.server_runner import seed_owner_state + seed_owner_state(data_dir) env = { **os.environ, "HOME": str(home), @@ -201,11 +211,14 @@ def test_group_sigterm_reaches_terminal_custody_without_a_false_supervisor_alarm "OUROBOROS_HOST_SERVICE_PORT": str(port + 1), "OUROBOROS_MANAGED_BY_LAUNCHER": "1", } + if launcher_pipe: + env["OUROBOROS_LAUNCHER_STOP_STDIN"] = "1" url = f"http://127.0.0.1:{port}" container = ProcessContainer() proc = container.spawn( [sys.executable, "server.py"], cwd=REPO_ROOT, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + stdin=subprocess.PIPE if launcher_pipe else subprocess.DEVNULL, ) ws = None stuck = None @@ -224,7 +237,12 @@ def test_group_sigterm_reaches_terminal_custody_without_a_false_supervisor_alarm ) time.sleep(0.5) started = time.monotonic() - os.killpg(os.getpgid(proc.pid), signal.SIGTERM) # exactly what the pre-#1142 launcher does + if launcher_pipe: + proc.stdin.write(b"quit\n") + proc.stdin.flush() + proc.stdin.close() + else: + os.killpg(os.getpgid(proc.pid), signal.SIGTERM) # pre-#1142 launcher try: code = proc.wait(timeout=LAUNCHER_STOP_GRACE_SEC) except subprocess.TimeoutExpired: @@ -249,12 +267,105 @@ def test_group_sigterm_reaches_terminal_custody_without_a_false_supervisor_alarm # uvicorn re-raises the captured SIGTERM once the lifespan has completed, so a clean # signalled exit reads -SIGTERM (or 0); -SIGKILL is the launcher fallback this test forbids. - assert code in (0, -signal.SIGTERM), f"graceful exit expected, got {code}" + expected_codes = (0,) if launcher_pipe else (0, -signal.SIGTERM) + assert code in expected_codes, f"graceful exit expected, got {code}" chat_rows = _read_rows(data_dir / "logs" / "chat.jsonl") alarms = [row for row in chat_rows if row.get("system_type") == "supervisor_failure" or "Supervisor loop died" in str(row.get("text") or "")] assert alarms == [], alarms shutdown_rows = [row for row in _read_rows(data_dir / "logs" / "supervisor.jsonl") if row.get("type") == "server_shutdown"] - assert shutdown_rows and shutdown_rows[-1].get("cause") == "external_signal", shutdown_rows + cause = "launcher_quit" if launcher_pipe else "external_signal" + assert shutdown_rows and shutdown_rows[-1].get("cause") == cause, shutdown_rows assert elapsed < LAUNCHER_STOP_GRACE_SEC, elapsed + + +@pytest.mark.parametrize("message,stopped", [(b"quit\n", True), (b"", False), (b"other", False)]) +def test_launcher_pipe_only_an_explicit_quit_stops_the_server(monkeypatch, message, stopped): + import io + import types + import uvicorn + from ouroboros import server_process as state + + started = [] + monkeypatch.setenv("OUROBOROS_LAUNCHER_STOP_STDIN", "1") + monkeypatch.setattr(state, "_stop_source", "external_signal") + monkeypatch.setattr(state.sys, "stdin", types.SimpleNamespace(buffer=io.BytesIO(message))) + class ImmediateThread: + def __init__(self, *, target, **kwargs): + self.target = target + def start(self): + started.append(True) + self.target() + monkeypatch.setattr(state.threading, "Thread", ImmediateThread) + state._exit_signalled.clear() + state._supervisor_stop.clear() + try: + server = state._SignalStopServer(uvicorn.Config(lambda scope, receive, send: None)) + server.watch_launcher_stop() + assert started == [True] + assert server.should_exit is stopped + assert state._exit_signalled.is_set() is stopped + assert state._supervisor_stop.is_set() is stopped + assert "OUROBOROS_LAUNCHER_STOP_STDIN" not in os.environ + finally: + state._exit_signalled.clear() + state._supervisor_stop.clear() + + +def test_direct_server_does_not_consume_stdin_without_launcher_opt_in(monkeypatch): + import uvicorn + from ouroboros import server_process as state + monkeypatch.delenv("OUROBOROS_LAUNCHER_STOP_STDIN", raising=False) + monkeypatch.setattr(state.threading, "Thread", lambda **kw: pytest.fail("unexpected stdin reader")) + server = state._SignalStopServer(uvicorn.Config(lambda scope, receive, send: None)) + server.watch_launcher_stop() + assert not server.should_exit + + +def test_windows_launcher_requests_cooperative_quit_before_wait(monkeypatch): + import launcher + calls = [] + class Pipe: + def write(self, data): calls.append(("write", data)) + def flush(self): calls.append("flush") + def close(self): calls.append("close") + class Process: + pid = 12345 + stdin = Pipe() + def terminate(self): pytest.fail("TerminateProcess must not be the graceful step") + def wait(self, timeout): calls.append(("wait", timeout)) + monkeypatch.setattr(launcher, "IS_WINDOWS", True) + monkeypatch.setattr(launcher, "_agent_proc", Process()) + monkeypatch.setattr(launcher, "_agent_job", None) + monkeypatch.setattr(launcher, "_cleanup_recorded_server_group_for_pid", lambda *args: None) + launcher.stop_agent() + assert calls == [("write", b"quit\n"), "flush", "close", ("wait", launcher.LAUNCHER_STOP_GRACE_SEC)] + + +@pytest.mark.parametrize("pipe_failure", [False, True]) +def test_windows_launcher_keeps_forced_fallback_when_grace_expires(monkeypatch, pipe_failure): + import launcher + calls = [] + class Pipe: + def write(self, data): + calls.append("request") + if pipe_failure: raise BrokenPipeError("fixture closed pipe") + def flush(self): pass + def close(self): + if pipe_failure: raise BrokenPipeError("fixture flush on close") + class Process: + pid = 12345 + stdin = Pipe() + def wait(self, timeout): + calls.append(("wait", timeout)) + if len([c for c in calls if isinstance(c, tuple)]) == 1: + raise subprocess.TimeoutExpired("server", timeout) + monkeypatch.setattr(launcher, "IS_WINDOWS", True) + monkeypatch.setattr(launcher, "_agent_proc", Process()) + monkeypatch.setattr(launcher, "_agent_job", "owned-job") + monkeypatch.setattr(launcher, "terminate_job", lambda job: calls.append("force")) + monkeypatch.setattr(launcher, "close_job", lambda job: calls.append("close_job")) + monkeypatch.setattr(launcher, "_cleanup_recorded_server_group_for_pid", lambda *args: None) + launcher.stop_agent() + assert calls == ["request", ("wait", launcher.LAUNCHER_STOP_GRACE_SEC), "force", ("wait", 5), "close_job"] diff --git a/tests/test_update_dirty_stash.py b/tests/test_update_dirty_stash.py index 39e86bb06..5916fc701 100644 --- a/tests/test_update_dirty_stash.py +++ b/tests/test_update_dirty_stash.py @@ -110,11 +110,11 @@ def test_stash_roundtrip_restores_local_work(tmp_path, monkeypatch): assert status == "ok" and stash_sha, error assert not _git(repo, "status", "--porcelain").stdout.strip() - restored, note = update_merge.restore_update_stash(stash_sha, context="test") - assert restored, note + result = update_merge.restore_update_stash(stash_sha, context="test") + assert result.status == "restored", result assert (repo / "a.txt").read_text() == "dirty\n" assert (repo / "untracked.txt").read_text() == "scratch\n" - assert not _git(repo, "stash", "list").stdout.strip() + assert stash_sha in _git(repo, "stash", "list", "--format=%H").stdout def test_stash_on_clean_tree_is_a_noop(tmp_path, monkeypatch): @@ -123,7 +123,7 @@ def test_stash_on_clean_tree_is_a_noop(tmp_path, monkeypatch): status, stash_sha, error = update_merge.stash_local_changes_for_update("t2") assert status == "ok" and stash_sha == "", (stash_sha, error) - assert update_merge.restore_update_stash("", context="test") == (True, "") + assert update_merge.restore_update_stash("", context="test").status == "not_needed" def test_conflicting_restore_keeps_stash_and_discloses(tmp_path, monkeypatch): @@ -137,10 +137,10 @@ def test_conflicting_restore_keeps_stash_and_discloses(tmp_path, monkeypatch): _git(repo, "add", "-A") _git(repo, "commit", "-q", "-m", "official rewrite") - restored, note = update_merge.restore_update_stash(stash_sha, context="test") + result = update_merge.restore_update_stash(stash_sha, context="test") - assert restored is False - assert stash_sha[:12] in note and "git stash apply" in note + assert result.status == "preserved" + assert stash_sha[:12] in result.note and "git stash apply" in result.note # Worktree left clean; the stash entry survives for manual recovery. assert not _git(repo, "status", "--porcelain").stdout.strip() assert stash_sha in _git(repo, "stash", "list", "--format=%H").stdout @@ -165,7 +165,7 @@ def test_boot_finalize_restores_stash_and_is_replay_safe(tmp_path, monkeypatch): assert result.get("finalized") is True, result assert (repo / "a.txt").read_text() == "owner work in flight\n" - assert not _git(repo, "stash", "list").stdout.strip() + assert stash_sha in _git(repo, "stash", "list", "--format=%H").stdout # Replay (e.g. double boot task): no tx left, restored content untouched. replay = update_merge.finalize_managed_update_on_boot(supervisor_ready=True) assert replay.get("finalized") is False diff --git a/tests/test_update_flow_composition.py b/tests/test_update_flow_composition.py index 6a7777d16..0d4c174cb 100644 --- a/tests/test_update_flow_composition.py +++ b/tests/test_update_flow_composition.py @@ -324,7 +324,7 @@ def test_dirty_work_survives_the_full_managed_clean_update(tmp_path, monkeypatch assert (repo / "a.txt").read_text() == "owner dirty work\n" assert (repo / "untracked.txt").read_text() == "scratch\n" assert (repo / "remote.txt").read_text() == "official\n" # the update landed - assert not tds._git(repo, "stash", "list").stdout.strip() + assert tx["stash_sha"] in tds._git(repo, "stash", "list", "--format=%H").stdout assert update_merge.read_update_tx() == {} diff --git a/tests/test_update_merge_assisted.py b/tests/test_update_merge_assisted.py index 8e47edf4a..5e047d800 100644 --- a/tests/test_update_merge_assisted.py +++ b/tests/test_update_merge_assisted.py @@ -315,40 +315,32 @@ def test_cleanup_only_gate_block_retries_the_marker_and_never_rolls_back(tmp_pat ) -def test_restore_skips_drop_when_the_stash_list_changed_mid_restore(tmp_path, monkeypatch): - """A concurrent stash push between the apply and the drop shifts every - selector: the restore must then KEEP the entry (litter) rather than drop a - selector that may now name someone else's work.""" +def test_restore_keeps_both_update_and_interleaved_foreign_stash(tmp_path, monkeypatch): + """No positional drop exists, so an external stash push cannot lose its entry.""" repo, head = _init_repo(tmp_path) _point_at(monkeypatch, tmp_path, repo, head) - (repo / "work.txt").write_text("ours\n") + (repo / "work.txt").write_text("ours\n", encoding="utf-8") status, our_sha, error = update_merge.stash_local_changes_for_update("race-test") assert status == "ok" and our_sha, error - real_capture = git_ops.git_capture - state = {"list_calls": 0} + calls = [] def racing_capture(cmd): - if cmd[:3] == ["git", "stash", "list"] and "--format=%H %gd" in cmd: - state["list_calls"] += 1 - if state["list_calls"] == 2: - # Interleave a foreign push right before the post-apply re-list. - (repo / "foreign.txt").write_text("someone else\n") - subprocess.run(["git", "-C", str(repo), "stash", "push", - "--include-untracked", "-m", "foreign", - "--", "foreign.txt"], - capture_output=True, text=True) - return real_capture(cmd) + calls.append(cmd) + result = real_capture(cmd) + if cmd[:3] == ["git", "stash", "apply"]: + (repo / "foreign.txt").write_text("someone else\n", encoding="utf-8") + assert _git(repo, "stash", "push", "--include-untracked", "-m", "foreign", + "--", "foreign.txt").returncode == 0 + return result monkeypatch.setattr(git_ops, "git_capture", racing_capture) - - restored, note = update_merge.restore_update_stash(our_sha, context="race") - - assert restored, note - assert (repo / "work.txt").read_text() == "ours\n" + result = update_merge.restore_update_stash(our_sha, context="race") + assert result.status == "restored", result + assert (repo / "work.txt").read_text(encoding="utf-8") == "ours\n" shas = _git(repo, "stash", "list", "--format=%H").stdout.split() - assert our_sha in shas, "our entry was dropped despite the shifted list" - assert len(shas) == 2 # the foreign entry survived too + assert our_sha in shas and len(shas) == 2 + assert not any(cmd[:3] == ["git", "stash", "drop"] for cmd in calls) def test_boot_backfill_reprojects_before_pinning_m0(tmp_path, monkeypatch): @@ -413,7 +405,7 @@ def test_restore_with_marker_refuses_a_dirty_tree(tmp_path, monkeypatch): (repo / "late.txt").write_text("late human edit\n") # tree dirty again tx = {"stash_sha": stash_sha} - note = update_merge.restore_stash_with_marker(tx, "unwind-test") + note = update_merge.restore_stash_with_marker(tx, "unwind-test").note assert "NOT auto-applied" in note and stash_sha[:12] in note assert (repo / "late.txt").read_text() == "late human edit\n" @@ -473,13 +465,13 @@ def test_live_unmerged_paths_error_is_not_no_conflicts(monkeypatch): from supervisor import update_candidate monkeypatch.setattr( - update_candidate._g, "git_capture", lambda cmd: (1, "", "boom") + update_candidate._g, "_run_git_process_bounded", lambda cmd, **kw: (1, b"", b"boom") ) assert update_candidate.live_unmerged_paths() is None def test_marker_guarded_restore_replay_does_not_wipe_restored_work(tmp_path, monkeypatch): - """Crash between the stash apply and its drop: the tx carries + """Crash after the stash apply was recorded: the tx carries stash_restored=True, so a replayed restore must be a no-op — never a conflicting re-apply whose cleanup resets the already-restored copy.""" repo, head = _init_repo(tmp_path) @@ -497,7 +489,7 @@ def test_marker_guarded_restore_replay_does_not_wipe_restored_work(tmp_path, mon # Simulate post-restore progress that a naive re-apply would clobber. (repo / "work.txt").write_text("owner work + more\n") note2 = update_merge.restore_stash_with_marker(tx, "replay") - assert note2 == "" # marker short-circuits: no re-apply, no reset + assert note2.status == "restored" # recorded outcome: no re-apply, no reset assert (repo / "work.txt").read_text() == "owner work + more\n", (note1, note2) diff --git a/tests/test_update_merge_plan.py b/tests/test_update_merge_plan.py index e69d71904..7557f8389 100644 --- a/tests/test_update_merge_plan.py +++ b/tests/test_update_merge_plan.py @@ -182,8 +182,8 @@ def test_build_and_apply_clean_merge(tmp_path, monkeypatch): ok, msg = update_merge.apply_managed_merge_update(head, plan["merge_commit"]) assert ok, msg assert (repo / "b.txt").exists() - restored, note = update_merge.restore_update_stash(stash_sha, context="test") - assert restored, note + result = update_merge.restore_update_stash(stash_sha, context="test") + assert result.status == "restored", result assert (repo / "c.txt").read_text() == "local untracked\n" # Base was fast-forwardable, so official history lands as-is: HEAD is the # target itself, with no synthetic merge commit carrying local work. diff --git a/tests/test_update_recovery_browser.py b/tests/test_update_recovery_browser.py new file mode 100644 index 000000000..bd19b2fc3 --- /dev/null +++ b/tests/test_update_recovery_browser.py @@ -0,0 +1,65 @@ +"""The Updates consumer must not offer Restart while local-work recovery holds it.""" +from functools import partial +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from threading import Thread + +import pytest +from tests.test_update_letter_browser import BOOT, WEB + +pytestmark = [pytest.mark.ui_browser, pytest.mark.serial] + + +@pytest.mark.parametrize('width', [1360, 390]) +def test_pending_restore_explains_real_server_restart(width, tmp_path): + if os.environ.get('OUROBOROS_RUN_UI_SMOKE') != '1': + pytest.skip('Set OUROBOROS_RUN_UI_SMOKE=1') + playwright = pytest.importorskip('playwright.sync_api') + status = {'managed': True, 'check_ok': True, 'available': False, + 'current_version': '7.5.1', 'current_sha': 'a' * 40, 'warnings': [], + 'update_tx': {'active': True, 'phase': 'pending_boot_smoke', 'local_work_recovery': True}} + class Handler(SimpleHTTPRequestHandler): + def log_message(self, *_args): pass + def do_GET(self): + if self.path == '/fixture': + self.send_response(200) + self.send_header('Content-Type', 'text/html; charset=utf-8') + self.end_headers() + html = (WEB / 'index.html').read_text(encoding='utf-8').replace( + '', BOOT) + self.wfile.write(html.encode('utf-8')) + else: + self.path = self.path.removeprefix('/static') + super().do_GET() + server = ThreadingHTTPServer(('127.0.0.1', 0), partial(Handler, directory=str(WEB))) + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + requests = [] + def respond(route): + requests.append((route.request.method, route.request.url)) + if '/api/update/' in route.request.url: + body = status + else: + body = {'commits': [], 'tags': []} + route.fulfill(content_type='application/json', body=json.dumps(body)) + try: + with playwright.sync_playwright() as pw: + browser = pw.chromium.launch(headless=True) + try: + page = browser.new_page(viewport={'width': width, 'height': 900}, has_touch=width < 980) + page.route('**/api/**', respond) + page.goto(f'http://127.0.0.1:{server.server_port}/fixture') + page.get_by_text('Local changes still need recovery confirmation.', exact=True).wait_for() + assert page.locator('#btn-update-primary').inner_text() == 'Check again' + assert 'restart the server process' in page.locator('#dashboard-panel-updates').inner_text() + page.locator('#btn-update-primary').click() + page.wait_for_function("document.querySelector('#btn-update-primary').textContent.trim() === 'Check again'") + assert not any('/api/restart' in url or '/api/update/apply' in url for _method, url in requests) + image = tmp_path / f'update-recovery-{width}.png' + page.screenshot(path=str(image), full_page=True) + print(f'UPDATE_RECOVERY_SCREENSHOT={image}') + finally: + browser.close() + finally: + server.shutdown(); server.server_close(); thread.join(timeout=5) diff --git a/tests/test_update_stash_recovery.py b/tests/test_update_stash_recovery.py new file mode 100644 index 000000000..5ae8fce8a --- /dev/null +++ b/tests/test_update_stash_recovery.py @@ -0,0 +1,402 @@ +"""Real-Git preservation and crash boundaries of the managed-update stash owner.""" + +import json + +import pytest + +from ouroboros.gateway import control +from supervisor import git_ops, update_candidate, update_merge +from tests.test_update_dirty_stash import _git, _init_repo, _point_at +from tests.test_update_merge_assisted import _stub_worker_gates + + +class Crash(BaseException): + """A process interruption must escape ordinary exception recovery.""" + + +def _fixture(tmp_path, monkeypatch, consumer="boot_finalize"): + repo, branch = _init_repo(tmp_path) + _point_at(monkeypatch, tmp_path, repo, branch) + _stub_worker_gates(monkeypatch) + pre = _git(repo, "rev-parse", "HEAD").stdout.strip() + (repo / "a.txt").write_text("owner work\n", encoding="utf-8") + (repo / "scratch.txt").write_text("owner scratch\n", encoding="utf-8") + status, sha, error = update_merge.stash_local_changes_for_update("fault-test") + assert status == "ok" and sha, error + assert update_merge.create_rescue_local_ref(sha) + tx = {"pre_update_sha": pre, "pre_update_branch": branch, "stash_sha": sha, + "target_sha": pre, "attempt_id": "fault-test"} + if consumer in {"boot_finalize", "boot_diverged", "rollback"}: + (repo / "official.txt").write_text("official\n", encoding="utf-8") + assert _git(repo, "add", "-A").returncode == 0 + assert _git(repo, "commit", "-qm", "official").returncode == 0 + current = _git(repo, "rev-parse", "HEAD").stdout.strip() + tx.update(target_sha=current, merge_commit=current, pre_restart_smoke="passed") + if consumer == "boot_diverged": + # An unrelated local descendant, not the update's merge commit. + tx.pop("merge_commit") + tx["target_sha"] = "f" * 40 + tx["phase"] = "assisted_resolution" + elif consumer == "rollback": + tx["phase"] = "rolling_back" + elif consumer == "boot_finalize": + tx["phase"] = "pending_boot_smoke" + else: + tx["phase"] = "stashing_local_work" + update_merge.write_update_tx(tx) + return repo, tx + + +def _drive(consumer, tx): + if consumer == "gateway_unwind": + return {"stash_note": control._unwind_stashed_update(tx, "test_unwind")} + if consumer == "rollback": + ok, note = update_merge.rollback_managed_update("test_rollback") + return {"rolled_back": ok, "stash_note": note} + return update_merge.finalize_managed_update_on_boot(supervisor_ready=True) + + +def _events(tmp_path): + path = tmp_path / "data" / "logs" / "supervisor.jsonl" + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] if path.exists() else [] + + +@pytest.mark.parametrize("consumer", ["rollback", "boot_finalize", "boot_diverged", "boot_preapply", "gateway_unwind"]) +def test_each_consumer_retains_failed_marker_and_replay_preserves_later_edits(tmp_path, monkeypatch, consumer): + repo, tx = _fixture(tmp_path, monkeypatch, consumer) + real_write = update_merge.write_update_tx + + def fail_final_marker(payload): + if payload.get("stash_restored"): + raise OSError("injected completion write failure") + real_write(payload) + + monkeypatch.setattr(update_merge, "write_update_tx", fail_final_marker) + result = _drive(consumer, tx) + assert "unconfirmed" in result["stash_note"], result + assert result.get("rolled_back") is not True + assert result.get("finalized") is not True + saved = update_merge.read_update_tx() + assert saved["stash_restore"]["status"] == "applying" + assert saved["_schema_version"] == 2 and not saved.get("stash_restored") + assert (repo / "a.txt").read_text(encoding="utf-8") == "owner work\n" + assert tx["stash_sha"] in _git(repo, "stash", "list", "--format=%H").stdout + assert not any(event["type"] == "managed_update_stash_restored" for event in _events(tmp_path)) + + (repo / "a.txt").write_text("owner work plus later edits\n", encoding="utf-8") + (repo / "later.txt").write_text("later untracked\n", encoding="utf-8") + monkeypatch.setattr(update_merge, "write_update_tx", real_write) + real_capture, calls = git_ops.git_capture, [] + + def capture(cmd): + calls.append(cmd) + return real_capture(cmd) + + monkeypatch.setattr(git_ops, "git_capture", capture) + resumed = update_merge.finalize_managed_update_on_boot(supervisor_ready=True) + assert "unconfirmed" in resumed["stash_note"] + assert update_merge.read_update_tx() == {} + assert (repo / "a.txt").read_text(encoding="utf-8") == "owner work plus later edits\n" + assert (repo / "later.txt").read_text(encoding="utf-8") == "later untracked\n" + assert not any(cmd[1] in {"reset", "clean", "checkout"} or cmd[:3] == ["git", "stash", "apply"] for cmd in calls) + + +def test_crash_between_apply_and_marker_uses_old_reader_refusal_then_new_reader_handoff(tmp_path, monkeypatch): + repo, tx = _fixture(tmp_path, monkeypatch, "rollback") + real_capture = git_ops.git_capture + + def crash_after_apply(cmd): + result = real_capture(cmd) + if cmd[:3] == ["git", "stash", "apply"]: + assert result[0] == 0 + raise Crash() + return result + + monkeypatch.setattr(git_ops, "git_capture", crash_after_apply) + with pytest.raises(Crash): + update_merge.rollback_managed_update("crash") + assert _git(repo, "rev-parse", "HEAD").stdout.strip() == tx["pre_update_sha"] + assert (repo / "a.txt").read_text(encoding="utf-8") == "owner work\n" + marker = update_merge._update_tx_marker_path() + before = marker.read_bytes() + monkeypatch.setattr(update_merge, "UPDATE_TX_SCHEMA_VERSION", 1) + monkeypatch.setattr(git_ops, "git_capture", lambda cmd: pytest.fail(f"old reader ran Git: {cmd}")) + assert update_merge.read_update_tx_strict()[0] == "future" + assert update_merge.rollback_managed_update("old-reader")[0] is False + refused = update_merge.finalize_managed_update_on_boot() + assert "newer version" in refused["reason"] + assert marker.read_bytes() == before + monkeypatch.setattr(update_merge, "UPDATE_TX_SCHEMA_VERSION", 2) + monkeypatch.setattr(git_ops, "git_capture", real_capture) + (repo / "a.txt").write_text("later owner work\n", encoding="utf-8") + resumed = update_merge.finalize_managed_update_on_boot() + assert resumed["rolled_back"] is True + assert "unconfirmed" in resumed["stash_note"] + assert (repo / "a.txt").read_text(encoding="utf-8") == "later owner work\n" + + +def test_rollback_checkout_before_restore_still_uses_schema_one(tmp_path, monkeypatch): + repo, tx = _fixture(tmp_path, monkeypatch, "rollback") + real_capture = git_ops.git_capture + + def crash_after_checkout(cmd): + result = real_capture(cmd) + if cmd[:3] == ["git", "checkout", "-B"]: + raise Crash() + return result + + monkeypatch.setattr(git_ops, "git_capture", crash_after_checkout) + with pytest.raises(Crash): + update_merge.rollback_managed_update("before-restore") + saved = update_merge.read_update_tx() + assert saved["_schema_version"] == 1 and "stash_restore" not in saved + monkeypatch.setattr(update_merge, "UPDATE_TX_SCHEMA_VERSION", 1) + assert update_merge.read_update_tx_strict()[0] == "valid" + monkeypatch.setattr(update_merge, "UPDATE_TX_SCHEMA_VERSION", 2) + monkeypatch.setattr(git_ops, "git_capture", real_capture) + assert update_merge.rollback_managed_update("resume")[0] is True + assert (repo / "a.txt").read_text(encoding="utf-8") == "owner work\n" + + +def test_write_ahead_failure_does_not_apply_or_clear_and_retry_still_restores(tmp_path, monkeypatch): + repo, tx = _fixture(tmp_path, monkeypatch, "boot_preapply") + real_write = update_merge.write_update_tx + monkeypatch.setattr(update_merge, "write_update_tx", lambda payload: (_ for _ in ()).throw(OSError("read-only disk"))) + result = update_merge.finalize_managed_update_on_boot() + assert result["reason"] == "stash_recovery_incomplete" + assert (repo / "a.txt").read_text(encoding="utf-8") == "base\n" + assert update_merge.read_update_tx()["stash_sha"] == tx["stash_sha"] + monkeypatch.setattr(update_merge, "write_update_tx", real_write) + result = update_merge.finalize_managed_update_on_boot() + assert result["stash_restore_status"] == "restored" + assert (repo / "a.txt").read_text(encoding="utf-8") == "owner work\n" + + +def _conflict(tmp_path, monkeypatch): + repo, branch = _init_repo(tmp_path) + _point_at(monkeypatch, tmp_path, repo, branch) + path = " conflict\nname.txt " + (repo / path).write_text("base\n", encoding="utf-8") + _git(repo, "add", "-A") + _git(repo, "commit", "-qm", "conflict base") + pre = _git(repo, "rev-parse", "HEAD").stdout.strip() + (repo / path).write_text("owner\n", encoding="utf-8") + (repo / "scratch.txt").write_text("owner scratch\n", encoding="utf-8") + status, sha, _ = update_merge.stash_local_changes_for_update("conflict") + assert status == "ok" + (repo / path).write_text("official\n", encoding="utf-8") + _git(repo, "add", "-A") + _git(repo, "commit", "-qm", "official") + target = _git(repo, "rev-parse", "HEAD").stdout.strip() + tx = {"phase": "pending_boot_smoke", "pre_update_sha": pre, "pre_update_branch": branch, + "target_sha": target, "merge_commit": target, "pre_restart_smoke": "passed", "stash_sha": sha} + update_merge.write_update_tx(tx) + return repo, tx, path + + +def test_real_conflict_census_and_stdout_are_recorded_before_cleanup_and_reach_notice(tmp_path, monkeypatch): + repo, tx, path = _conflict(tmp_path, monkeypatch) + real_capture = git_ops.git_capture + before_cleanup = [] + + def capture(cmd): + if cmd[:3] == ["git", "reset", "--hard"]: + record = update_merge.read_update_tx()["stash_restore"] + assert record["status"] == "cleanup_pending" + assert record["conflict_paths"] == [path] + assert record["apply_returncode"] == 1 + assert record["stdout"] and not record["stderr"] + assert (repo / "scratch.txt").exists() + before_cleanup.append(record) + return real_capture(cmd) + + monkeypatch.setattr(git_ops, "git_capture", capture) + result = update_merge.finalize_managed_update_on_boot() + assert before_cleanup and result["finalized"] is True + assert result["stash_restore_status"] == "preserved" + assert json.dumps(path) in result["stash_note"] + assert "Git diagnostics:" in result["stash_note"] + assert "git stash apply " + tx["stash_sha"] in result["stash_note"] + assert _git(repo, "status", "--porcelain").stdout == "" + assert tx["stash_sha"] in _git(repo, "stash", "list", "--format=%H").stdout + event = next(row for row in _events(tmp_path) if row["type"] == "managed_update_stash_restore_failed") + assert event["conflict_paths"] == [path] and event["cleanup"]["verified_clean"] is True + assert event["stdout"] == before_cleanup[0]["stdout"] + assert not update_merge.active_update_tx() + + +def test_cleanup_failure_is_not_success_and_replay_preserves_unmerged_files(tmp_path, monkeypatch): + from supervisor import worker_chat_lane, workers + + monkeypatch.setattr(workers, "_repo_writer_gate_reason", "") + repo, tx, path = _conflict(tmp_path, monkeypatch) + real_capture = git_ops.git_capture + + def fail_reset(cmd): + if cmd[:3] == ["git", "reset", "--hard"]: + return 1, "", "index write denied" + return real_capture(cmd) + + monkeypatch.setattr(git_ops, "git_capture", fail_reset) + result = update_merge.finalize_managed_update_on_boot() + assert result["finalized"] is False and "cleanup failed" in result["stash_note"] + assert update_candidate.live_unmerged_paths() == [path] + partial = (repo / path).read_bytes() + (repo / "late.txt").write_text("later work\n", encoding="utf-8") + monkeypatch.setattr(git_ops, "git_capture", real_capture) + resumed = update_merge.finalize_managed_update_on_boot() + assert resumed["stash_restore_status"] == "preserved" + assert "unconfirmed" in resumed["stash_note"] and "cleanup failed" in resumed["stash_note"] + assert (repo / path).read_bytes() == partial + assert (repo / "late.txt").read_text(encoding="utf-8") == "later work\n" + assert update_candidate.live_unmerged_paths() == [path] + assert update_merge.active_update_tx() == {} # normal owner repair is available + assert workers.repo_writer_admission_closed() == "" + assert worker_chat_lane.owner_conversation_admitted(1) is True + + +def test_missing_stash_list_entry_uses_pinned_object_and_missing_object_stays_incomplete(tmp_path, monkeypatch): + repo, tx = _fixture(tmp_path, monkeypatch, "boot_preapply") + assert _git(repo, "stash", "drop", "stash@{0}").returncode == 0 + result = update_merge.finalize_managed_update_on_boot() + assert result["stash_restore_status"] == "restored" + assert (repo / "a.txt").read_text(encoding="utf-8") == "owner work\n" + assert _git(repo, "rev-parse", "rescue-local-" + tx["stash_sha"][:12]).stdout.strip() == tx["stash_sha"] + update_merge.write_update_tx({**tx, "stash_sha": "f" * 40}) + missing = update_merge.finalize_managed_update_on_boot() + assert missing["reason"] == "stash_recovery_incomplete" + assert update_merge.active_update_tx() + assert (repo / "a.txt").read_text(encoding="utf-8") == "owner work\n" + + +def test_event_append_failure_retries_only_disclosure(tmp_path, monkeypatch): + repo, tx = _fixture(tmp_path, monkeypatch, "boot_preapply") + real_log = update_merge._log_supervisor + monkeypatch.setattr(update_merge, "_log_supervisor", lambda row: False) + failed = update_merge.finalize_managed_update_on_boot() + assert failed["reason"] == "stash_recovery_incomplete" + assert update_merge.read_update_tx()["stash_restored"] is True + (repo / "a.txt").write_text("later edits\n", encoding="utf-8") + monkeypatch.setattr(update_merge, "_log_supervisor", real_log) + monkeypatch.setattr(git_ops, "git_capture", lambda cmd: pytest.fail(f"disclosure replay ran Git: {cmd}")) + resumed = update_merge.finalize_managed_update_on_boot() + assert resumed["stash_restore_status"] == "restored" + assert (repo / "a.txt").read_text(encoding="utf-8") == "later edits\n" + assert not update_merge.active_update_tx() + + +def test_secondary_completion_event_failure_keeps_the_deliverable_note(tmp_path, monkeypatch): + repo, tx = _fixture(tmp_path, monkeypatch, "boot_preapply") + real_append = update_merge.append_jsonl + + def append(path, row, **kwargs): + if row["type"] == "managed_update_stash_recovered_on_boot": + raise OSError("secondary event failure") + return real_append(path, row, **kwargs) + + monkeypatch.setattr(update_merge, "append_jsonl", append) + result = update_merge.finalize_managed_update_on_boot() + assert result["stash_restore_status"] == "restored" + assert "update stash was retained" in result["stash_note"] + assert not update_merge.active_update_tx() + assert (repo / "a.txt").read_text(encoding="utf-8") == "owner work\n" + assert any(row["type"] == "managed_update_stash_restored" for row in _events(tmp_path)) + + +@pytest.mark.parametrize("census", [[], None]) +def test_failed_apply_without_census_does_not_invent_conflict_names(tmp_path, monkeypatch, census): + repo, tx = _fixture(tmp_path, monkeypatch, "boot_preapply") + real_capture = git_ops.git_capture + monkeypatch.setattr(update_candidate, "live_unmerged_paths", lambda: census) + + def fail_apply(cmd): + return (1, "apply output", "apply failure") if cmd[:3] == ["git", "stash", "apply"] else real_capture(cmd) + + monkeypatch.setattr(git_ops, "git_capture", fail_apply) + result = update_merge.finalize_managed_update_on_boot() + assert result["stash_restore_status"] == "preserved" + assert "apply output" in result["stash_note"] and "apply failure" in result["stash_note"] + assert ("inventory could not be read" if census is None else "without tracked conflicts") in result["stash_note"] + event = next(row for row in _events(tmp_path) if row["type"] == "managed_update_stash_restore_failed") + assert event["conflict_paths"] == census + + +def test_gateway_exception_after_restore_does_not_enter_destructive_rollback(tmp_path, monkeypatch): + repo, tx = _fixture(tmp_path, monkeypatch, "gateway_unwind") + real_active = update_merge.active_update_tx + first_check = [True] + + def active(): + if first_check: + first_check.pop() + return {} + return real_active() + + monkeypatch.setattr(update_merge, "active_update_tx", active) + monkeypatch.setattr(update_merge, "clear_update_tx", lambda: False) + monkeypatch.setattr(control, "_quiesce_repo_writers", lambda reason: []) + monkeypatch.setattr(control, "_stash_local_work_fenced", lambda **kw: (tx, None)) + monkeypatch.setattr(control, "_rollback_fenced_update", lambda *a, **kw: pytest.fail("destructive generic rollback")) + plan = {"available": True, "kind": "clean", "base_sha": tx["pre_update_sha"], + "target_sha": tx["target_sha"]} + + def plan_then_fail(**kwargs): + if kwargs.get("build"): + control._unwind_stashed_update(tx, "exception_test") + raise RuntimeError("response delivery failed") + return plan + + monkeypatch.setattr(update_merge, "plan_managed_update_merge", plan_then_fail) + response = control._apply_smart_update_fenced( + object(), expected_base_sha=tx["pre_update_sha"], expected_target_sha=tx["target_sha"], + ) + body = json.loads(response.body) + assert response.status_code == 500 and body["reason"] == "stash_recovery_incomplete" + assert "stash" in body["stash_note"] and body["restart_required"] is True + assert (repo / "a.txt").read_text(encoding="utf-8") == "owner work\n" + assert real_active()["stash_restored"] is True + + +@pytest.mark.parametrize("ok", [True, False]) +def test_gateway_rollback_disclosure_uses_the_actual_stash_note_consumer(monkeypatch, ok): + note = "rolled back to abc; conflicting paths: draft.txt; recover with git stash apply def" + monkeypatch.setattr(update_merge, "rollback_managed_update", lambda reason: (ok, note)) + monkeypatch.setattr(update_merge, "active_update_tx", lambda: {}) + monkeypatch.setattr(update_merge, "mark_update_tx_gate_blocked", lambda *a: None) + monkeypatch.setattr(control, "_respawn_workers_after_failed_update", lambda: None) + response = control._rollback_fenced_update("test", "smoke failed") + body = json.loads(response.body) + assert body["rolled_back"] is ok + assert body["stash_note"] == note == body["rollback"] + + +@pytest.mark.parametrize("restore_status", [None, "applying", "restored", "preserved"]) +def test_restart_keeps_ordinary_boot_smoke_but_never_resets_pending_restore(tmp_path, monkeypatch, restore_status): + from ouroboros.server_restart import _safe_restart_serialized + + repo, tx = _fixture(tmp_path, monkeypatch) + if restore_status: + # Both the uncertain window and a recorded restore awaiting event/tx + # cleanup may hold returned work. The current reader must refuse too. + tx["stash_restore"] = {"status": restore_status, "stash_sha": tx["stash_sha"]} + update_merge.write_update_tx(tx) + assert _git(repo, "stash", "apply", tx["stash_sha"]).returncode == 0 + (repo / "a.txt").write_text("returned work plus later edits\n", encoding="utf-8") + before = (repo / "a.txt").read_bytes() + calls = [] + + def destructive_restart(**kwargs): + calls.append(kwargs) + assert _git(repo, "reset", "--hard", "HEAD").returncode == 0 + assert _git(repo, "clean", "-fd").returncode == 0 + return True, "restarted" + + ok, message = _safe_restart_serialized( + destructive_restart, reason="owner_restart", unsynced_policy="rescue_and_reset", + ) + if restore_status: + assert ok is False and not calls + assert "Quit and reopen" in message and "server process" in message + assert (repo / "a.txt").read_bytes() == before + else: + assert ok is True and calls == [{"reason": "owner_restart", "unsynced_policy": "rescue_and_reset"}] diff --git a/tests/test_update_status_cache.py b/tests/test_update_status_cache.py index 866705422..bb13ba9b7 100644 --- a/tests/test_update_status_cache.py +++ b/tests/test_update_status_cache.py @@ -292,3 +292,15 @@ def test_fetching_payload_writes_the_letter_through_the_one_seam(monkeypatch): assert calls and calls[0]["fetched"] is True assert payload["letter"] is None and payload["latest_version"] == "6.87.5" + + +def test_pending_local_work_is_projected_without_exposing_raw_transaction(monkeypatch): + import supervisor.update_merge as update_merge + _wire(monkeypatch) + tx = {"phase": "pending_boot_smoke", "stash_restore": {"status": "applying", "stash_sha": "s" * 40}} + monkeypatch.setattr(update_merge, "active_update_tx", lambda: tx) + payload = control._managed_update_payload(fetch=False, include_tags=False) + assert payload["update_tx"]["local_work_recovery"] is True + assert "stash_restore" not in payload["update_tx"] + tx.pop("stash_restore") + assert "local_work_recovery" not in control._managed_update_payload(fetch=False, include_tags=False)["update_tx"] diff --git a/tests/test_update_tx_nminus1_shim.py b/tests/test_update_tx_nminus1_shim.py index fef386147..f7ca2928c 100644 --- a/tests/test_update_tx_nminus1_shim.py +++ b/tests/test_update_tx_nminus1_shim.py @@ -56,7 +56,7 @@ def test_write_update_tx_stamps_without_mutating_the_caller(tmp_path, monkeypatc update_merge.write_update_tx(payload) assert SCHEMA_VERSION_KEY not in payload # never mutate the caller's dict raw = _raw_marker() - assert raw[SCHEMA_VERSION_KEY] == update_merge.UPDATE_TX_SCHEMA_VERSION + assert raw[SCHEMA_VERSION_KEY] == 1 status, tx = update_merge.read_update_tx_strict() assert status == "valid" and tx["task_id"] == "x" @@ -213,7 +213,7 @@ def test_n_minus_1_assisted_resolution_resumes_and_upgrades_the_marker(tmp_path, assert enqueued and enqueued[0]["task_id"] == "resolver" assert (repo / "a.txt").read_text() == "the resolver's precious resolution\n" raw = _raw_marker() # the resume re-write stamped the surviving marker - assert raw[SCHEMA_VERSION_KEY] == update_merge.UPDATE_TX_SCHEMA_VERSION + assert raw[SCHEMA_VERSION_KEY] == 1 assert raw["phase"] == "assisted_resolution" diff --git a/tests/test_worker_readiness.py b/tests/test_worker_readiness.py index 00c114620..80371b9e8 100644 --- a/tests/test_worker_readiness.py +++ b/tests/test_worker_readiness.py @@ -4,8 +4,8 @@ What is pinned, on fake process objects (no child is ever forked here): * spawn installs every slot ``reaping`` and hands the whole set to the ONE readiness seam; * respawn installs its fresh slot the same way, through the same seam, carrying its attempt count; -* the seam opens a slot only on the child's OWN ``worker_ready`` row (matched by pid), verifying the - booted SHA in the same step; a foreign pid's row does not open it; +* the seam opens a slot only on the child's OWN ``worker_ready`` row (matched by pid); its checkout + SHA comparison is diagnostic and never changes admission; a foreign pid's row does not open it; * no ``worker_ready`` inside the window -> the child is torn down (process tree), the slot is replaced through ``respawn_worker`` and a typed ``worker_ready_timeout`` row names slot, pid, wait and reason; its own entry progress permits one extension to the birth-relative ceiling; @@ -30,6 +30,7 @@ from __future__ import annotations import json import logging import os +import subprocess import time from types import SimpleNamespace from unittest.mock import MagicMock @@ -211,6 +212,7 @@ def test_the_slot_opens_only_on_its_own_worker_ready_row_and_verifies_the_sha(po verify = _rows(seam.supervisor, "worker_sha_verify") assert len(verify) == 1 assert verify[0]["ok"] is True and verify[0]["worker_id"] == 0 and verify[0]["worker_pid"] == 5001 + assert verify[0]["relation"] == "equal" assert verify[0]["slot_opened"] is True and verify[0]["attempt"] == 1 assert _rows(seam.supervisor, "worker_ready_timeout") == [] @@ -326,16 +328,126 @@ def test_a_late_watcher_cannot_grant_a_fresh_window_after_the_ceiling(pool, seam assert _rows(seam.supervisor, "worker_ready_timeout")[0]["waited_sec"] == 4.0 -def test_sha_mismatch_on_the_ready_row_opens_the_slot_and_tells_the_owner(pool, seam, monkeypatch): - monkeypatch.setattr(pool.workers, "load_state", lambda: {"current_sha": "abc123", "owner_chat_id": 7}) +@pytest.fixture +def sha_history(pool): + """A real graph: root -> baseline -> descendant, plus root -> other branch.""" + def git(*args): + return subprocess.run( + ["git", *args], cwd=pool.root, check=True, capture_output=True, text=True, + timeout=10, + ).stdout.strip() + + def commit(name): + git("-c", "user.name=Worker SHA Test", "-c", "user.email=worker@example.test", + "-c", "commit.gpgsign=false", "commit", "--allow-empty", "-m", name) + return git("rev-parse", "HEAD") + + git("init", "-b", "main") + root = commit("root") + baseline = commit("baseline") + descendant = commit("descendant") + git("checkout", "-b", "other", root) + other = commit("other") + return {"root": root, "baseline": baseline, "descendant": descendant, "other": other, + "missing": "f" * 40, "absent": ""} + + +@pytest.mark.parametrize("expected,observed,relation,warn", [ + ("baseline", "baseline", "equal", False), + ("baseline", "descendant", "descendant", False), + ("baseline", "root", "non_descendant", True), + ("baseline", "other", "non_descendant", True), + ("baseline", "missing", "unavailable", True), + ("missing", "descendant", "unavailable", True), + ("baseline", "absent", "unavailable", True), + ("absent", "descendant", "not_applicable", False), + ("absent", "absent", "not_applicable", False), +]) +def test_checkout_relation_is_a_fact_not_a_worker_admission_gate( + pool, seam, monkeypatch, sha_history, expected, observed, relation, warn, +): + expected_sha, observed_sha = sha_history[expected], sha_history[observed] + state = {"current_sha": expected_sha, "owner_chat_id": 7} + monkeypatch.setattr(pool.workers, "load_state", lambda: dict(state)) + from supervisor import state as supervisor_state + update_state = MagicMock() + monkeypatch.setattr(supervisor_state, "update_state", update_state) slot = _booting_slot(pool, 0, 5001) - append_jsonl(seam.events, {"type": "worker_ready", "worker_id": 0, "pid": 5001, "git_sha": "other"}) + append_jsonl(seam.events, {"type": "worker_ready", "worker_id": 0, "pid": 5001, + "git_sha": observed_sha}) + real_run, queries = subprocess.run, [] + + def observe_query(command, **kwargs): + assert slot.reaping is False, "ready capacity opens before diagnostic I/O" + assert command == (["git", "rev-parse", "--is-shallow-repository"] if queries else + ["git", "merge-base", "--is-ancestor", expected_sha, observed_sha]) + assert kwargs["cwd"] == pool.root + assert kwargs["timeout"] == pool.lifecycle._WORKER_SHA_LOOKUP_TIMEOUT_SEC + assert kwargs["env"]["GIT_NO_LAZY_FETCH"] == "1" + queries.append(command) + return real_run(command, **kwargs) + + monkeypatch.setattr(pool.lifecycle.subprocess, "run", observe_query) seam.run({0: slot}, seam.cursor, 1) assert slot.reaping is False - assert _rows(seam.supervisor, "worker_sha_verify")[0]["ok"] is False - assert seam.sent and "SHA mismatch" in seam.sent[0][1] + assert seam.killed == [] and seam.respawned == [] + row, = _rows(seam.supervisor, "worker_sha_verify") + assert row["relation"] == relation + assert row["expected_sha"] == expected_sha and row["observed_sha"] == observed_sha + assert row["ok"] is ((bool(observed_sha) and expected_sha == observed_sha) if expected_sha else None) + assert row["slot_opened"] is True + assert len(queries) == (2 if relation == "non_descendant" else + int(bool(expected_sha and observed_sha and expected_sha != observed_sha))) + assert bool(seam.sent) is warn + if warn: + assert len(seam.sent) == 1 and seam.sent[0][0] == 7 + assert expected_sha[:8] in seam.sent[0][1] and (observed_sha or "unknown")[:8] in seam.sent[0][1] + assert ("not a descendant" if relation == "non_descendant" else "could not be compared") in seam.sent[0][1] + assert bool(_rows(seam.supervisor, "worker_sha_verify_skipped")) is (not expected_sha) + update_state.assert_not_called() + + +@pytest.mark.parametrize("boundary,relation", [("descendant", "unavailable"), ("root", "descendant")]) +def test_shallow_history_keeps_proven_ancestry_but_cannot_prove_non_ancestry(pool, seam, monkeypatch, sha_history, boundary, relation): + baseline, descendant = sha_history["baseline"], sha_history["descendant"] + # A real Git shallow boundary hides descendant's parent traversal without + # deleting baseline's object. The same graph above proves ancestry when full. + (pool.root / ".git" / "shallow").write_text(sha_history[boundary] + "\n", encoding="utf-8") + monkeypatch.setattr(pool.workers, "load_state", lambda: {"current_sha": baseline, "owner_chat_id": 7}) + slot = _booting_slot(pool, 0, 5001) + append_jsonl(seam.events, {"type": "worker_ready", "pid": 5001, "git_sha": descendant}) + + seam.run({0: slot}, seam.cursor, 1) + + assert slot.reaping is False and seam.killed == [] and seam.respawned == [] + row, = _rows(seam.supervisor, "worker_sha_verify") + assert row["ok"] is False and row["relation"] == relation + if relation == "unavailable": + assert row["relation_error"] == "shallow_history" + assert len(seam.sent) == 1 and "could not be compared" in seam.sent[0][1] + else: + assert seam.sent == [] + + +@pytest.mark.parametrize("error", [OSError("git unavailable"), subprocess.TimeoutExpired("git", 2)]) +def test_checkout_lookup_failure_is_unavailable_without_holding_ready_capacity(pool, seam, monkeypatch, error): + monkeypatch.setattr(pool.workers, "load_state", lambda: {"current_sha": "a" * 40, "owner_chat_id": 7}) + slot = _booting_slot(pool, 0, 5001) + append_jsonl(seam.events, {"type": "worker_ready", "pid": 5001, "git_sha": "b" * 40}) + query = MagicMock(side_effect=error) + monkeypatch.setattr(pool.lifecycle.subprocess, "run", query) + + seam.run({0: slot}, seam.cursor, 1) + + assert slot.reaping is False and seam.killed == [] and seam.respawned == [] + query.assert_called_once() + row, = _rows(seam.supervisor, "worker_sha_verify") + assert row["ok"] is False and row["relation"] == "unavailable" + assert row["relation_error_type"] == type(error).__name__ + assert len(seam.sent) == 1 and "could not be compared" in seam.sent[0][1] + assert _rows(seam.supervisor, "worker_ready_released") == [] def test_no_worker_ready_inside_the_window_tears_down_replaces_and_types_the_row(pool, seam): diff --git a/tests/test_worker_sha_browser.py b/tests/test_worker_sha_browser.py new file mode 100644 index 000000000..375a8099c --- /dev/null +++ b/tests/test_worker_sha_browser.py @@ -0,0 +1,77 @@ +"""Worker checkout observations in the real Logs renderer, desktop and phone.""" +from functools import partial +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from threading import Thread +import os + +import pytest + +pytestmark = [pytest.mark.ui_browser, pytest.mark.serial] +WEB = Path(__file__).resolve().parents[1] / 'web' +BOOT = '''''' + + +@pytest.mark.parametrize('width', [1360, 390]) +def test_descendant_is_informational_and_problem_stays_warning(width, tmp_path): + if os.environ.get('OUROBOROS_RUN_UI_SMOKE') != '1': + pytest.skip('Set OUROBOROS_RUN_UI_SMOKE=1') + playwright = pytest.importorskip('playwright.sync_api') + class Handler(SimpleHTTPRequestHandler): + def log_message(self, *_args): pass + def do_GET(self): + if self.path == '/fixture': + self.send_response(200) + self.send_header('Content-Type', 'text/html; charset=utf-8') + self.end_headers() + html = (WEB / 'index.html').read_text(encoding='utf-8').replace( + '', BOOT) + self.wfile.write(html.encode('utf-8')) + else: + self.path = self.path.removeprefix('/static') + super().do_GET() + server = ThreadingHTTPServer(('127.0.0.1', 0), partial(Handler, directory=str(WEB))) + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + with playwright.sync_playwright() as pw: + browser = pw.chromium.launch(headless=True) + try: + page = browser.new_page(viewport={'width': width, 'height': 900}, has_touch=width < 980) + errors = [] + page.on('pageerror', lambda exc: errors.append(str(exc))) + page.route('**/api/**', lambda route: route.fulfill(json={'entries': []})) + page.goto(f'http://127.0.0.1:{server.server_port}/fixture') + page.wait_for_function('typeof window.emitWorker === "function"') + page.evaluate("emitWorker('descendant', 'b', 101)") + page.evaluate("emitWorker('non_descendant', 'c', 102)") + page.evaluate("emitWorker('unavailable', 'd', 103)") + descendant = page.locator('.log-entry').filter(has_text='Worker checkout descends from baseline') + assert descendant.locator('.log-phase').inner_text() == 'info' + problem = page.locator('.log-entry').filter(has_text='Worker checkout is not a descendant of baseline') + assert problem.locator('.log-phase').inner_text() == 'warn' + assert page.locator('.log-entry').filter(has_text='comparison unavailable').locator('.log-phase').inner_text() == 'warn' + descendant.get_by_role('button', name='Raw', exact=True).click() + assert 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb' in descendant.locator('.log-raw').inner_text() + assert 'Worker SHA mismatch' not in page.locator('#page-logs').inner_text() + assert not errors, errors + page.screenshot(path=str(tmp_path / f'worker-sha-{width}.png'), full_page=True) + print(f'WORKER_SHA_SCREENSHOT={tmp_path / f"worker-sha-{width}.png"}') + finally: + browser.close() + finally: + server.shutdown(); server.server_close(); thread.join(timeout=5) diff --git a/web/modules/log_events.js b/web/modules/log_events.js index 97dd4f025..ea987ec83 100644 --- a/web/modules/log_events.js +++ b/web/modules/log_events.js @@ -8,6 +8,7 @@ import { fmt, tr } from './i18n.js'; import { historyRetentionView } from './history_retention.js'; import { effortEvidenceText } from './effort_evidence.js'; import { delegatedActivityView } from './delegated_activity.js'; +import { workerShaLogView, workerShaRelation } from './worker_sha_presentation.js'; import { classifyReviewLifecycle, classifyReviewLifecyclePointer, @@ -1023,13 +1024,8 @@ export function summarizeLogEvent(evt) { evt.ceiling_sec != null ? `ceiling ${evt.ceiling_sec}s` : ''] }); if (t === 'worker_sha_verify') { - return view(evt.ok ? 'ok' : 'warn', evt.ok ? 'Worker SHA verified' : 'Worker SHA mismatch', { - meta: [ - evt.expected_sha ? `exp ${String(evt.expected_sha).slice(0, 8)}` : '', - evt.observed_sha ? `got ${String(evt.observed_sha).slice(0, 8)}` : '', - evt.worker_pid ? `pid ${evt.worker_pid}` : '', - ], - }); + const result = workerShaLogView(evt); + return view(result.phase, result.headline, result); } if (t === 'worker_boot') return view('boot', 'Worker booted', { @@ -1534,7 +1530,7 @@ function summarizeChatLiveEventView(evt) { export function duplicateLogEventKey(evt) { const t = evt.type || evt.event || ''; if (t === 'startup_verification') return `${t}:${evt.git_sha || ''}:${evt.issues_count || 0}`; - if (t === 'worker_sha_verify') return `${t}:${evt.expected_sha || ''}:${evt.observed_sha || ''}:${evt.ok ? 1 : 0}`; + if (t === 'worker_sha_verify') return `${t}:${evt.expected_sha || ''}:${evt.observed_sha || ''}:${workerShaRelation(evt)}`; if (t === 'deps_sync_ok') return `${t}:${evt.reason || ''}:${evt.source || ''}`; return ''; } diff --git a/web/modules/updates.js b/web/modules/updates.js index 4969899ca..6c0874c38 100644 --- a/web/modules/updates.js +++ b/web/modules/updates.js @@ -82,6 +82,24 @@ function progressVerdict(data, base) { return null; // Assisted work and boot recovery have their existing durable owner. } +export function applyFailureText(err) { + const body = err?.body || {}; + const parts = [String(err.message || err)]; + if (body.reason) parts.push(`reason: ${body.reason}`); + if (Array.isArray(body.blockers) && body.blockers.length) { + parts.push(`blocked by: ${body.blockers.slice(0, 5).join(', ')}`); + } + if (body.rolled_back) parts.push('the checkout was rolled back'); + if (body.smoke) parts.push('the post-update smoke check failed'); + if (typeof body.estimated_wave_usd === 'number') { + parts.push(`assisted review needs ~$${body.estimated_wave_usd}` + ( + typeof body.remaining_usd === 'number' ? ` of $${body.remaining_usd} remaining` : '' + )); + } + if (body.stash_note) parts.push(body.stash_note); + return parts.join(' · '); +} + // Verdict function: durable server state × transient client phase → one // presentation descriptor (deterministic given status, phase, and the clock — // humanizeCheckedAt reads Date.now for the "checked N ago" age). The button is always a real next action; facts @@ -102,6 +120,13 @@ export function updateVerdict(data = {}, phase = '') { const warnings = extraWarnings(data); const base = { chips, warnings, checkedAgo }; + if (data.update_tx?.active && data.update_tx.local_work_recovery) { + return { ...base, state: 'resolving', tone: 'warn', + headline: 'Local changes still need recovery confirmation.', + hint: 'Quit and reopen the desktop app, or restart the server process for a web deployment. ' + + 'In-app Restart is deferred to preserve the current files. The saved update stash is retained.', + action: { id: 'check', label: 'Check again' } }; + } const recovery = data.update_tx?.active && ( ['corrupt', 'gate_blocked', 'marker_cleanup_retry'].includes(data.update_tx.phase) || (!data.update_progress?.active && data.update_progress?.result === 'failed') @@ -855,24 +880,6 @@ export function initUpdates({ mount, state, ws, openSettingsTab }) { } } - function applyFailureText(err) { - const body = err?.body || {}; - const parts = [String(err.message || err)]; - if (body.reason) parts.push(`reason: ${body.reason}`); - if (Array.isArray(body.blockers) && body.blockers.length) { - parts.push(`blocked by: ${body.blockers.slice(0, 5).join(', ')}`); - } - if (body.rolled_back) parts.push('the checkout was rolled back'); - if (body.smoke) parts.push('the post-update smoke check failed'); - if (typeof body.estimated_wave_usd === 'number') { - parts.push(`assisted review needs ~$${body.estimated_wave_usd}` + ( - typeof body.remaining_usd === 'number' ? ` of $${body.remaining_usd} remaining` : '' - )); - } - if (body.stash_note) parts.push(body.stash_note); - return parts.join(' · '); - } - async function applyUpdate() { if (!latestStatus?.available) return; setPhase('preflighting'); diff --git a/web/modules/worker_sha_presentation.js b/web/modules/worker_sha_presentation.js new file mode 100644 index 000000000..1bc89b1e3 --- /dev/null +++ b/web/modules/worker_sha_presentation.js @@ -0,0 +1,43 @@ +// Checkout observations only: never authorship, review or loaded-module proof. +// Historical rows carry equality facts but cannot gain ancestry retrospectively. +import { fmt, tr } from './i18n.js'; + +export function workerShaRelation(evt) { + if (evt.relation) return String(evt.relation); + if (!evt.expected_sha) return 'not_applicable'; + if (!evt.observed_sha) return 'unavailable'; + return evt.expected_sha === evt.observed_sha ? 'equal' : 'unrecorded'; +} + +export function workerShaLogView(evt) { + let phase = 'info', headline, body = ''; + switch (workerShaRelation(evt)) { + case 'equal': + phase = 'ok'; + headline = tr('worker.sha.equal', 'Worker checkout matches baseline'); + break; + case 'descendant': + headline = tr('worker.sha.descendant', 'Worker checkout descends from baseline'); + break; + case 'non_descendant': + phase = 'warn'; + headline = tr('worker.sha.non_descendant', 'Worker checkout is not a descendant of baseline'); + break; + case 'unavailable': + phase = 'warn'; + headline = tr('worker.sha.unavailable', 'Worker checkout comparison unavailable'); + break; + case 'not_applicable': + headline = tr('worker.sha.not_applicable', 'Worker checkout comparison skipped'); + body = tr('worker.sha.no_baseline', 'No managed baseline recorded.'); + break; + default: + headline = tr('worker.sha.different', 'Worker checkout differs from baseline'); + body = tr('worker.sha.unrecorded', 'Ancestry was not recorded for this observation.'); + } + return { phase, headline, body, meta: [ + evt.expected_sha ? fmt('baseline {sha}', { sha: String(evt.expected_sha).slice(0, 8) }) : '', + evt.observed_sha ? fmt('observed {sha}', { sha: String(evt.observed_sha).slice(0, 8) }) : '', + evt.worker_pid ? `pid ${evt.worker_pid}` : '', + ] }; +} diff --git a/web/tests/update_flow.test.js b/web/tests/update_flow.test.js index 266772431..35976bec2 100644 --- a/web/tests/update_flow.test.js +++ b/web/tests/update_flow.test.js @@ -5,11 +5,31 @@ import { readFileSync } from 'node:fs'; import { apiClient, updateStrategyForPlan } from '../modules/api_client.js'; import { updatePillText, verifiedUpdatePlan } from '../modules/update_status.js'; import { + applyFailureText, + updateVerdict, bindUpdateRefreshEvents, restartStatusCanSettle, } from '../modules/updates.js'; +test('update failure renders the rollback stash disclosure without losing diagnostics', () => { + const stash = 'a'.repeat(40); + const note = `rolled back to 123456789abc; conflicting paths: "draft\\nname.txt"; ` + + `recover with \`git stash apply ${stash}\`\nGit diagnostics:\nCONFLICT from stdout`; + const rendered = applyFailureText({ message: 'smoke failed', body: { + rolled_back: true, rollback: note, stash_note: note, + } }); + assert.ok(rendered.includes(note)); + assert.equal(rendered.split(`git stash apply ${stash}`).length - 1, 1); + const pending = applyFailureText({ message: 'write failed', body: { + reason: 'stash_recovery_incomplete', rolled_back: false, + stash_note: 'Automatic recovery is unconfirmed; current files were left unchanged.', + } }); + assert.ok(pending.includes('unconfirmed')); + assert.ok(!pending.includes('the checkout was rolled back')); +}); + + class FakeWS { constructor() { this.listeners = new Map(); } on(type, handler) { @@ -260,3 +280,17 @@ test('all successful restart paths share one lifecycle entry', () => { assert.equal((source.match(/enterRestarting\(\)/g) || []).length, 5); assert.doesNotMatch(source, /restartNeeded = false;\s*enterRestarting\(\)/); }); + + +test('pending local-work recovery never offers the refused in-app Restart', () => { + const ordinary = { update_tx: { active: true, phase: 'pending_boot_smoke' } }; + assert.equal(updateVerdict(ordinary).action.id, 'restart'); + const pending = { update_tx: { ...ordinary.update_tx, local_work_recovery: true } }; + for (const phase of ['', 'restarting', 'restart_needed', 'restart_required']) { + const view = updateVerdict(pending, phase); + assert.equal(view.action.id, 'check'); + assert.match(view.hint, /Quit and reopen/); + assert.match(view.hint, /restart the server process/); + assert.match(view.hint, /stash is retained/); + } +}); diff --git a/web/tests/worker_sha_presentation.test.js b/web/tests/worker_sha_presentation.test.js new file mode 100644 index 000000000..fe2022594 --- /dev/null +++ b/web/tests/worker_sha_presentation.test.js @@ -0,0 +1,74 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { summarizeLogEvent, duplicateLogEventKey, categorizeLogEvent } from '../modules/log_events.js'; +import { applyPayload, CODE_PREFIX } from '../modules/i18n.js'; + +const observation = { + type: 'worker_sha_verify', expected_sha: 'a'.repeat(40), observed_sha: 'b'.repeat(40), + ok: false, worker_pid: 123, +}; + +test('Logs renders relation facts without turning inequality into a warning', () => { + const cases = [ + ['equal', 'ok', 'Worker checkout matches baseline'], + ['descendant', 'info', 'Worker checkout descends from baseline'], + ['non_descendant', 'warn', 'Worker checkout is not a descendant of baseline'], + ['unavailable', 'warn', 'Worker checkout comparison unavailable'], + ['not_applicable', 'info', 'Worker checkout comparison skipped'], + ]; + for (const [relation, phase, headline] of cases) { + const event = { ...observation, relation }; + if (relation === 'equal') { event.observed_sha = event.expected_sha; event.ok = true; } + if (relation === 'not_applicable') { event.expected_sha = ''; event.ok = null; } + const view = summarizeLogEvent(event); + assert.equal(view.phase, phase, relation); + assert.equal(view.headline, headline, relation); + assert.equal(categorizeLogEvent(event), 'system'); + assert.ok(view.meta.includes('pid 123')); + assert.equal(event.ok, relation === 'equal' ? true : relation === 'not_applicable' ? null : false); + } +}); + +test('historical rows retain only their recorded equality facts', () => { + const different = summarizeLogEvent(observation); + assert.equal(different.phase, 'info'); + assert.equal(different.headline, 'Worker checkout differs from baseline'); + assert.match(different.body, /Ancestry was not recorded/); + assert.deepEqual(different.meta, ['baseline aaaaaaaa', 'observed bbbbbbbb', 'pid 123']); + const equal = summarizeLogEvent({ ...observation, observed_sha: observation.expected_sha, ok: true }); + assert.equal(equal.phase, 'ok'); + assert.equal(equal.headline, 'Worker checkout matches baseline'); + const skipped = summarizeLogEvent({ ...observation, expected_sha: '', ok: null }); + assert.equal(skipped.phase, 'info'); + assert.equal(skipped.headline, 'Worker checkout comparison skipped'); + assert.equal(skipped.body, 'No managed baseline recorded.'); + assert.equal(summarizeLogEvent({ ...observation, observed_sha: '' }).phase, 'warn'); +}); + +test('repeat grouping cannot hide a changed or missing ancestry observation', () => { + const variants = [observation, ...['descendant', 'non_descendant', 'unavailable'].map( + (relation) => ({ ...observation, relation }), + )]; + assert.equal(new Set(variants.map(duplicateLogEventKey)).size, variants.length); + const descendant = { ...observation, relation: 'descendant' }; + assert.equal(duplicateLogEventKey(descendant), duplicateLogEventKey({ ...descendant, worker_pid: 456 })); + assert.notEqual(duplicateLogEventKey(descendant), duplicateLogEventKey({ ...descendant, observed_sha: 'c'.repeat(40) })); +}); + +test('new host labels use the translation memory and leave SHA values intact', () => { + applyPayload({ language: 'ru', english: false, revision: 1, entries: { + [CODE_PREFIX + 'worker.sha.descendant']: { text: 'Коммит рабочего процесса продолжает базовую историю' }, + 'baseline {sha}': { text: 'база {sha}' }, + 'observed {sha}': { text: 'наблюдается {sha}' }, + } }); + try { + const event = { ...observation, relation: 'descendant' }; + const before = duplicateLogEventKey(event); + const view = summarizeLogEvent(event); + assert.equal(view.headline, 'Коммит рабочего процесса продолжает базовую историю'); + assert.deepEqual(view.meta, ['база aaaaaaaa', 'наблюдается bbbbbbbb', 'pid 123']); + assert.equal(duplicateLogEventKey(event), before); + } finally { + applyPayload({ language: '', english: true, revision: 0, entries: {} }); + } +});