Tell the agent where its salvaged work went (#3684)

docs/architecture/orchestrator.md

@@ -1044,14 +1044,16 @@ section is the wait-side companion to this architecture description.
 ### Per-event prompt composer (`compose_event_prompt`)
 
 `compose_event_prompt(role, event_payload, memory_excerpt, nacks,
-git_log_delta, base_branch, *, task_description="", iteration_feedback=None) -> str` builds the
+git_log_delta, base_branch, *, task_description="", iteration_feedback=None,
+release_context=None, recovery_context=None, ...) -> str` builds the
 one-shot user prompt the wrapper invokes the agent with at each
 `case action.kind == INVOKE`
 branch of the deterministic loop (see [Deterministic loop structure](#deterministic-loop-structure)
 above). It assembles, in order:
 
 | Position | Section | Source | Bound |
 |----------|---------|--------|-------|
+| **First** (ahead of the banner) | Worktree-recovery notice — "your previous session's work was PRESERVED, not lost" (#3684) | `EGG_WORKTREE_RECOVERY` env, set by `kubernetes_spawner._events.spawn_event_job` only on the spawn that follows a re-attach discard; decoded by `_cli.py` and rendered by `_render_recovery_section` | Env payload ≤ `_WORKTREE_RECOVERY_ENV_MAX_BYTES` (2 KiB); trailing per-repo entries drop first, and a lone oversized notice degrades to `_WORKTREE_RECOVERY_MINIMAL_FIELDS` (ref + shas) rather than to nothing. The one free-text field (`salvage_error`) is clamped to `_SALVAGE_ERROR_NOTICE_MAX_CHARS` (400) at the producer and whitespace-collapsed at the renderer — it is remote-controlled `git push` stderr. Omitted (byte-identical common path) on every ordinary spawn. Leads the *whole* prompt, ahead of the event banner: an agent that reads the event, opens the tree, and finds its files missing has already concluded "I must re-implement" by the time a later section could correct it. |
 | Top | Role banner + one-line event description | `role` + `event_payload.kind` | A few hundred bytes; identifies the producer/reviewer side of the dispatch. |
 | Top | Park-release delta — "why you were respawned" (#3537) | `EGG_EVENT_RELEASE_CONTEXT` env, set only on the probe spawn a no-op-park fingerprint-change release granted; decoded by `_cli.py` and rendered by `_render_release_context_section` | Each free-text field (question/resolution) ≤ `RELEASE_RESOLUTION_MAX_CHARS` (1200 chars); omitted (byte-identical common path) on every ordinary spawn and on heartbeat releases. See [No-op park release delta](#deterministic-loop-structure) above. |
 | Top | Task & operator directives (#3123/#3163) | The contract's `task_description`, read from the worktree contract file via the pod-inherited `EGG_PIPELINE_ID` / `EGG_ISSUE_NUMBER`; populated for all pipeline types since #3163 (issue anchor + submit description); omitted only when the contract carries no task statement and no issue identity | ≤ 4 KB (`TASK_DESCRIPTION_MAX_CHARS`), truncated with a pointer to `mcp__sdlc__show_contract`. Pushes the operator's submit-time directives into every invocation instead of relying on the agent pulling them per the rules file. |

docs/reference/agent-recovery.md

@@ -307,6 +307,15 @@ Agent restart preserves the agent's git worktree. The gateway's `create_worktree
 
 **Committed work is preserved only conditionally.** `_clean_reused_worktree` keeps local commits when the worktree is already at the `origin/<branch>` tip, or when the tree was clean **and** strictly ahead of the tip ([#3506](https://github.com/jwbron/egg/issues/3506)). Otherwise — notably whenever the tree was dirty — the unpushed commits are enumerated as orphans, pushed to an `egg/recovered/…` ref, and hard-reset away, so the respawned agent starts at the tip and its prior commits are reachable only via that ref.
 
+**The respawned agent is told where its work went** ([#3684](https://github.com/jwbron/egg/issues/3684)). Preserving the commits is only half the job: an agent whose session memory expired sees a worktree at the tip with its files gone, and without a pointer its only rational move is to re-implement. That happened — a coder re-derived 3072 insertions that were sitting on a recovery ref the whole time. Two channels now carry the ref:
+
+- **Push (the one that reaches an agent that is not looking).** The re-attach hands each discard's `{recovery_ref, tip_sha, reset_to, n_commits, fast_forward, …}` back to `spawn_event_job`, which injects it as `EGG_WORKTREE_RECOVERY` on the successor pod. `routes/event_prompt` renders it as the **first** section of that pod's prompt, ahead of the event banner. Set only on the spawn that follows a discard; every ordinary spawn leaves the key unset.
+- **Pull (the durable record).** The [#3509](https://github.com/jwbron/egg/issues/3509) bus record, now written with `Message.phase` set. Both readers behind `mcp__brc__read_peer_artifact` — the live `/brc-transcript` route and `_write_brc_history` — select on `m.message_type in BRC_HISTORY_TYPES and m.phase == phase`, so the record's `phase=None` default had made it unreadable through the only bus channel an agent has: written every time, readable never.
+
+Both channels name the restore command the gateway will actually run. `git reset --hard <recovery-tip>` is refused: pipeline sessions enforce an off-lineage-reset guard (`gateway/gateway/_git_execute.py`) requiring the target be an *ancestor* of HEAD, and a recovery tip is a descendant by construction. When the doomed tip was a strict descendant of the reset target the instruction is `git fetch origin <ref>` + `git merge --ff-only <sha>` (allowed, unguarded, loses nothing); on a diverged tip it is `git cherry-pick <reset_to>..<sha>`. The message says "preserved", never "discarded" — the old subject line, `Unpushed commits discarded on re-attach`, is what the incident's agent read and believed.
+
+**Not auto-restored.** A fast-forwardable tip could in principle be replayed without an agent round trip, but the R6 residue policy hard-resets a dirty pre-reset tree on purpose: the successor must not inherit a killed-mid-event working set. Reversing that is a policy change, tracked separately on [#3684](https://github.com/jwbron/egg/issues/3684).
+
 **`restart_agent_job` is retained but unreached.** It calls `auto_salvage_pipeline(..., salvage_uncommitted=True)`, which writes a `[salvage] pre-crash working-tree state (#2807)` snapshot ([#2807](https://github.com/jwbron/egg/issues/2807) / [#2855](https://github.com/jwbron/egg/pull/2855)). Nothing in production has called it since [#3164](https://github.com/jwbron/egg/issues/3164) — the restart budget it used to enforce in-band was moved into the route for that reason. When triaging a restart, grep the recovery refs for `pre-reset`, not `pre-crash`.
 
 **Implementation detail:** `spawn_agent_container()` always calls the gateway to create (or reuse) the per-agent worktree when `repos` is provided, regardless of whether `repo_volumes` was passed by the caller. This ensures both the initial spawn path and the restart path (which does not pass `repo_volumes`) correctly mount the agent's worktree. See issue [#1597](https://github.com/jwbron/egg/issues/1597) for the fix that resolved a bug where the restart path skipped worktree creation.

integration_tests/regression/test_unpushed_commit_salvage.py

@@ -898,3 +898,86 @@ def test_cleanup_pipeline_pushes_one_ref_per_worktree(self, tmp_path):
             "issue-2659-coder",
             "issue-2659-slice-3-coder",
         }.issubset(deleted_ids)
+
+
+# ---------------------------------------------------------------------------
+# #3684 — the record has to survive the filters its readers apply
+# ---------------------------------------------------------------------------
+
+
+class TestDiscardRecordSurvivesTheAgentChannel:
+    """A written record an agent cannot read is not a record (#3684).
+
+    ``mcp__brc__read_peer_artifact`` is the ONLY bus channel an agent has,
+    and both of its sources — the live ``/brc-transcript`` route and the
+    on-disk ``brc-history`` files ``_write_brc_history`` writes — select
+    with ``m.message_type in BRC_HISTORY_TYPES and m.phase == phase``.
+    ``_record_discarded_tip`` was the one STATUS emitter that left
+    ``Message.phase`` at its ``None`` default, so its record matched
+    neither filter: the salvage ran, the ref existed, and the agent that
+    needed it was told nothing.
+
+    This pins the coupling from both ends — that the emitter sets a phase,
+    and that the filter is really what turns that into visibility — so a
+    future change to either side cannot silently re-open the gap.
+    """
+
+    _PHASE = "implement"
+
+    def _record(self, **overrides):
+        from kubernetes_spawner._worktree import _record_discarded_tip
+
+        kwargs = {
+            "pipeline_id": "pipe-1",
+            "agent_worktree_id": "pipe-1-slice-1-coder",
+            "repo": "egg",
+            "branch": "egg/issue-1/slice-1",
+            "agent_role": "coder",
+            "slice_id": "slice-1",
+            "phase": self._PHASE,
+            "discarded_tip": "a" * 40,
+            "remote_tip": "b" * 40,
+            "n_commits": 8,
+            "was_dirty": True,
+            "recovery_ref": "egg/recovered/pipe-1/slice-1-coder/aaaaaaaaaaaa",
+            "salvage_error": None,
+            **overrides,
+        }
+        with patch("message_store.get_message_store") as get_store:
+            _record_discarded_tip(**kwargs)
+        return get_store.return_value.add_message.call_args.args[0]
+
+    @staticmethod
+    def _transcript_filter(messages, phase):
+        """The exact selection both readers apply."""
+        from routes.pipelines import BRC_HISTORY_TYPES
+
+        return [m for m in messages if m.message_type in BRC_HISTORY_TYPES and m.phase == phase]
+
+    def test_record_is_visible_to_the_transcript_filter(self):
+        msg = self._record()
+        assert self._transcript_filter([msg], self._PHASE) == [msg]
+
+    def test_a_phaseless_record_is_dropped_by_the_same_filter(self):
+        """The pre-#3684 behaviour, pinned as the thing that must not return."""
+        msg = self._record(phase=None)
+        assert self._transcript_filter([msg], self._PHASE) == []
+
+    def test_record_keeps_the_slice_scope_the_history_writer_buckets_on(self):
+        """``_write_brc_history`` partitions implement-phase records by
+        ``metadata['slice_id']``; without it the record lands in the
+        cross-cutting ``unattributed`` file instead of the slice's own."""
+        from slice_id_validation import SLICE_ID_PATTERN
+
+        msg = self._record()
+        assert SLICE_ID_PATTERN.fullmatch(msg.metadata["slice_id"])
+
+    def test_the_ref_and_a_usable_restore_command_are_in_the_body(self):
+        """The incident's agent had the ref two sentences away and still
+        re-derived, because the body led with "discarded" and the command
+        it offered (``reset --hard``) is one the gateway 403s."""
+        msg = self._record(ff_restorable=True)
+        assert "egg/recovered/pipe-1/slice-1-coder/aaaaaaaaaaaa" in msg.body
+        assert f"git merge --ff-only {'a' * 40}" in msg.body
+        assert "Do NOT `git reset --hard`" in msg.body
+        assert "PRESERVED" in msg.subject

orchestrator/kubernetes_spawner/__init__.py

@@ -196,6 +196,44 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
 ENV_EVENT_DEDUPE_KEY = "EGG_EVENT_DEDUPE_KEY"
 ENV_EVENT_PAYLOAD_REFS = "EGG_EVENT_PAYLOAD_REFS"
 
+# Recovery notice for work this spawn's worktree re-attach just moved off the
+# tree and onto an ``egg/recovered/...`` ref (#3684). JSON list, one entry per
+# repo that hit the discard path; set ONLY on the spawn that follows such a
+# discard, so every ordinary spawn renders a byte-identical prompt. Read by
+# ``routes/event_prompt/_cli.py`` and rendered as the leading prompt section —
+# the push half of the delivery, complementing the ``Message.phase``-gated bus
+# record that an agent has to go looking for. Same shape and same contract as
+# #3537's ``EGG_EVENT_RELEASE_CONTEXT``.
+ENV_WORKTREE_RECOVERY = "EGG_WORKTREE_RECOVERY"
+
+# Byte budget for ``ENV_WORKTREE_RECOVERY``. The composer's whole prompt
+# envelope is 10 KiB (``PROMPT_ENVELOPE_MAX_BYTES``); these entries are a
+# fixed set of shas and counts, so the cap only ever bites on a pathological
+# multi-repo discard, where dropping the tail beats dropping the ref.
+_WORKTREE_RECOVERY_ENV_MAX_BYTES = 2048
+
+# Clamp for the one free-text field in a recovery notice. ``salvage_error`` is
+# ``PushResult.describe()``, whose ``detail`` is raw ``git push`` stderr — every
+# ``remote:`` line the server echoes, including pre-receive hook output. That is
+# routinely multi-KiB, and an unbounded value in a 2 KiB budget is not a "drop
+# the tail" case: the notice is a singleton, so overshoot used to empty the list
+# and deliver nothing at all — on the salvage-FAILED arm, the one arm where the
+# agent most needs to hear "do NOT re-derive" (#3689 review). The full string
+# survives in the WARNING log and in the bus record's metadata, so clamping here
+# costs the operator nothing.
+_SALVAGE_ERROR_NOTICE_MAX_CHARS = 400
+
+# Fields kept when a single notice still will not fit the env budget after the
+# clamp above. The ref and the two shas ARE the payload — everything else is
+# elaboration — so a degraded notice beats no notice.
+_WORKTREE_RECOVERY_MINIMAL_FIELDS = (
+    "repo",
+    "recovery_ref",
+    "tip_sha",
+    "reset_to",
+    "fast_forward",
+)
+
 # A short, deterministic Job-name discriminator so distinct events for one
 # role get distinct Job names (the same event always yields the same name,
 # which keeps the pre-spawn cleanup + adoption coherent). 8 hex chars of the
@@ -512,6 +550,7 @@ def get_kubernetes_spawner(
 )
 from ._models import SpawnedContainer, _EventJobStatusView  # noqa: E402
 from ._worktree import (  # noqa: E402
+    _clamp_salvage_error,
     _host_to_local_volumes,
     _load_local_mount_mapping,
     _local_to_host_path,
@@ -537,6 +576,7 @@ def get_kubernetes_spawner(
 # through the barrel like the rest of the private surface.
 _job_is_live = _events._job_is_live
 _job_is_terminating = _events._job_is_terminating
+_recovery_env_json = _events._recovery_env_json
 KubernetesSpawner.stop_agent_job = _jobs.stop_agent_job
 KubernetesSpawner.remove_agent_job = _jobs.remove_agent_job
 # Module-level (not a class method) so ``remove_agent_job`` reaches it via the
@@ -571,6 +611,7 @@ def get_kubernetes_spawner(
     "get_kubernetes_spawner",
     "WORKTREE_BASE_DIR",
     "LABEL_EVENT_DEDUPE",
+    "ENV_WORKTREE_RECOVERY",
     "_PROTECTED_ENV_KEYS",
     "_ROLES_WITHOUT_WORKTREE",
     "ContainerInfo",
@@ -587,6 +628,8 @@ def get_kubernetes_spawner(
     "_dedupe_label_value",
     "_job_is_live",
     "_job_is_terminating",
+    "_recovery_env_json",
+    "_clamp_salvage_error",
     "_forwarded_discipline_env",
     "_resolve_live_phase",
     "_resolve_wait_producer_allowlist",