fix(engine): preserve prompt cache prefix with skill hook context

CHANGELOG.md

@@ -28,6 +28,8 @@
 
 ### Changed
 
+- Anchor skill-hook context in request-only batches within each turn, preserving the growing prompt-cache prefix
+  after compaction without separating tool calls from results.
 - With no `provider` configured, the agent uses the first provider with credentials, in the order openai, anthropic,
   gemini, openai-chatgpt, github-copilot, and that provider's default model. With no `model` configured, the model is
   the provider's default. An environment with an Anthropic key and no OpenAI key resolves as before. See

docs/concepts/skills.md

@@ -116,7 +116,11 @@ also includes `tool_response`. Post-tool hooks do not run for a call the caller
 Stop hooks run after successful model/tool work; cancellation starts no new hook work.
 
 Plain stdout or JSON `hookSpecificOutput.additionalContext` supplies context for later
-model requests in that turn. A nonzero exit, a timeout, a malformed result,
+model requests in that turn. Context batches keep their first request position until
+compaction, so later messages extend the prompt prefix. A batch waits until pending
+tool calls have results, and expires when the turn completes, fails, or is cancelled.
+This request-only context does not become a synthetic transcript message.
+A nonzero exit, a timeout, a malformed result,
 `continue: false`, `decision: block`, or `permissionDecision: deny` fails the turn with
 `tool_failed`. An allowing hook cannot override host approval. Denying a hook command's
 approval ends the turn `rejected` with `approval_denied`, since the work it guards cannot

packages/engine/src/amplifier_agent_engine/_engine/adapters.py

@@ -200,6 +200,11 @@ def __init__(self, tool_result_max_bytes: int | None, *, hooks: Any = None) -> N
         )
         # The compaction stats object the notice describes, its index in the view, and the message.
         self._notice_anchor: tuple[dict[str, Any], int, dict[str, Any]] | None = None
+        # Request-only skill context is anchored in batches so later conversation
+        # growth extends, rather than changes, the provider-visible prompt prefix.
+        self._skill_context_anchors: list[tuple[int, dict[str, Any]]] = []
+        self._skill_context_count = 0
+        self._skill_context_stats: dict[str, Any] | None = None
         # The input budget the engine sets for the selection in use, or None for the provider's window.
         self.model_budget: Callable[[], int | None] = lambda: None
         self.entry_token: str | None = None
@@ -247,9 +252,36 @@ async def get_messages_for_request(
     ) -> list[dict[str, Any]]:
         messages = await super().get_messages_for_request(token_budget=token_budget, provider=provider)
         messages = self._with_compaction_notice(messages)
-        if self.skill_context:
-            messages.append({"role": "user", "content": "\n\n".join(self.skill_context)})
-        return messages
+        return self._with_skill_context(messages)
+
+    def _with_skill_context(self, view: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """Insert turn-scoped hook context without changing an earlier request prefix."""
+        stats = self._last_compaction_stats
+        if self._skill_context_count > len(self.skill_context) or (
+            self._skill_context_anchors and self._skill_context_stats is not stats
+        ):
+            self._reset_skill_context_anchors()
+
+        result = list(view)
+        for index, message in self._skill_context_anchors:
+            if index > len(result):
+                self._reset_skill_context_anchors()
+                return self._with_skill_context(view)
+            result.insert(index, copy.deepcopy(message))
+
+        additions = self.skill_context[self._skill_context_count :]
+        if additions and not self._tail_has_unanswered_tool_calls(result):
+            message = {"role": "user", "content": "\n\n".join(additions)}
+            self._skill_context_anchors.append((len(result), message))
+            self._skill_context_count = len(self.skill_context)
+            self._skill_context_stats = stats
+            result.append(copy.deepcopy(message))
+        return result
+
+    def _reset_skill_context_anchors(self) -> None:
+        self._skill_context_anchors = []
+        self._skill_context_count = 0
+        self._skill_context_stats = None
 
     def _with_compaction_notice(self, view: list[dict[str, Any]]) -> list[dict[str, Any]]:
         """Insert the notice for the latest compaction at the position it was first placed.
@@ -279,10 +311,12 @@ def _with_compaction_notice(self, view: list[dict[str, Any]]) -> list[dict[str,
 
     async def set_messages(self, messages: list[dict[str, Any]]) -> None:
         self._notice_anchor = None
+        self._reset_skill_context_anchors()
         await super().set_messages(messages)
 
     async def clear(self) -> None:
         self._notice_anchor = None
+        self._reset_skill_context_anchors()
         await super().clear()
 
     def _derive_budget(self, token_budget: int | None, provider: Any | None) -> int:
@@ -941,6 +975,7 @@ async def execute(self, input: TurnInput, observer: Observer) -> None:
         self.observer = observer
         self.skill_hooks.begin(self.inherited_skill_hooks)
         self.context.skill_context.clear()
+        self.context._reset_skill_context_anchors()
         self._call_ids.clear()
         self.core.coordinator.cancellation.reset()
         self._turn_start = len(await self.context.get_messages())
@@ -966,6 +1001,7 @@ async def execute(self, input: TurnInput, observer: Observer) -> None:
         finally:
             self.skill_hooks.clear()
             self.context.skill_context.clear()
+            self.context._reset_skill_context_anchors()
             self.context.turn_active = False
             self.observer = None
 

packages/engine/tests/test_compaction_notice.py

@@ -136,3 +136,152 @@ async def test_an_anchor_beyond_the_view_moves_to_its_end():
     view = context._with_compaction_notice([{"role": "system", "content": "s"}, {"role": "user", "content": "u"}])
     assert notices(view) == [2]
     assert wire([view[2]]) == wire([anchor[2]])
+
+
+async def test_skill_context_preserves_request_prefix_within_turn():
+    context = await filled_context()
+    context.skill_context = ["PreToolUse context"]
+
+    first = await context.get_messages_for_request(token_budget=BUDGET)
+    stats = context._last_compaction_stats
+    assert stats is not None
+
+    await context.add_message({"role": "assistant", "content": "Continuing."})
+    await context.add_message({"role": "user", "content": "Next action."})
+
+    second = await context.get_messages_for_request(token_budget=BUDGET)
+
+    assert context._last_compaction_stats is stats
+    assert wire(second[: len(first)]) == wire(first)
+    stored = await context.get_messages()
+    assert not any("PreToolUse context" in str(message) for message in stored)
+    assert not any(is_notice(message) for message in stored)
+
+
+async def test_additional_skill_hook_context_keeps_previous_request_stable():
+    context = await filled_context()
+    context.skill_context = ["First hook context"]
+
+    first = await context.get_messages_for_request(token_budget=BUDGET)
+    stats = context._last_compaction_stats
+    assert stats is not None
+
+    await context.add_message({"role": "assistant", "content": "Tool completed."})
+    await context.add_message({"role": "user", "content": "Continue."})
+
+    context.skill_context.append("Second hook context")
+    second = await context.get_messages_for_request(token_budget=BUDGET)
+
+    assert context._last_compaction_stats is stats
+    assert wire(second[: len(first)]) == wire(first)
+    assert any("Second hook context" in str(message) for message in second)
+    assert not any("First hook context" in str(message) for message in await context.get_messages())
+    assert not any("Second hook context" in str(message) for message in await context.get_messages())
+
+
+async def test_skill_context_waits_for_an_unanswered_tool_call():
+    context = await filled_context()
+    await tool_round(context, 99, answered=False)
+    context.skill_context = ["PreToolUse context"]
+
+    view = await context.get_messages_for_request(token_budget=BUDGET)
+    assert view[-1]["tool_calls"][0]["id"] == "call-99"
+    assert not any("PreToolUse context" in str(message) for message in view)
+
+    await context.add_message({"role": "tool", "name": "read_file", "tool_call_id": "call-99", "content": "done"})
+    view = await context.get_messages_for_request(token_budget=BUDGET)
+    call_index = next(
+        index
+        for index, message in enumerate(view)
+        if any(call["id"] == "call-99" for call in message.get("tool_calls", []))
+    )
+    assert view[call_index + 1]["tool_call_id"] == "call-99"
+    assert view[-1] == {"role": "user", "content": "PreToolUse context"}
+
+
+async def test_skill_batches_reanchor_once_after_a_second_compaction():
+    context = await filled_context()
+    context.skill_context = ["first hook"]
+    await context.get_messages_for_request(token_budget=BUDGET)
+    context.skill_context.append("second hook")
+    await context.add_message({"role": "user", "content": "Continue."})
+    await context.get_messages_for_request(token_budget=BUDGET)
+    stats = context._last_compaction_stats
+    for index in range(12, 24):
+        await tool_round(context, index)
+    stored = copy.deepcopy(await context.get_messages())
+    compacted = await context.get_messages_for_request(token_budget=BUDGET)
+    assert context._last_compaction_stats is not stats
+    assert compacted[-1] == {"role": "user", "content": "first hook\n\nsecond hook"}
+    assert sum("first hook" in str(message) for message in compacted) == 1
+    assert sum("second hook" in str(message) for message in compacted) == 1
+    assert notices(compacted) == [len(compacted) - 2]
+    assert await context.get_messages() == stored
+    await context.add_message({"role": "user", "content": "One more."})
+    extended = await context.get_messages_for_request(token_budget=BUDGET)
+    assert wire(extended[: len(compacted)]) == wire(compacted)
+
+
+async def test_skill_context_reanchors_after_history_replacement_and_clear():
+    context = await filled_context()
+    context.skill_context = ["active hook"]
+    await context.get_messages_for_request(token_budget=BUDGET)
+    replacement = [{"role": "user", "content": "Replacement history.", "metadata": {}}]
+    await context.set_messages(copy.deepcopy(replacement))
+    stored = copy.deepcopy(await context.get_messages())
+    view = await context.get_messages_for_request(token_budget=BUDGET)
+    assert view == [*replacement, {"role": "user", "content": "active hook"}]
+    assert await context.get_messages() == stored
+    await context.clear()
+    view = await context.get_messages_for_request(token_budget=BUDGET)
+    assert view == [{"role": "user", "content": "active hook"}]
+    assert await context.get_messages() == []
+    await context.add_message({"role": "user", "content": "Fresh history."})
+    extended = await context.get_messages_for_request(token_budget=BUDGET)
+    assert wire(extended[: len(view)]) == wire(view)
+
+
+async def test_cleared_skill_context_cannot_replay_an_old_batch():
+    context = await filled_context()
+    context.skill_context = ["expired hook"]
+    await context.get_messages_for_request(token_budget=BUDGET)
+    context.skill_context.clear()
+    without_hook = await context.get_messages_for_request(token_budget=BUDGET)
+    assert "expired hook" not in wire(without_hook)
+    context.skill_context.append("new hook")
+    view = await context.get_messages_for_request(token_budget=BUDGET)
+    assert view[-1] == {"role": "user", "content": "new hook"}
+    assert wire(view[: len(without_hook)]) == wire(without_hook)
+
+
+async def test_skill_context_request_mutation_does_not_change_later_requests():
+    context = await filled_context()
+    context.skill_context = ["original hook"]
+    view = await context.get_messages_for_request(token_budget=BUDGET)
+    expected = copy.deepcopy(view)
+    view[-1]["content"] = "mutated by a request consumer"
+    assert await context.get_messages_for_request(token_budget=BUDGET) == expected
+
+
+async def test_new_hook_batch_waits_for_all_parallel_tool_results():
+    context = StructuredContext(131_072)
+    await context.add_message({"role": "user", "content": "Start."})
+    context.skill_context = ["earlier hook"]
+    first = await context.get_messages_for_request(token_budget=BUDGET)
+    await context.add_message(
+        {
+            "role": "assistant",
+            "content": "Read both.",
+            "tool_calls": [{"id": name, "tool": "read_file", "arguments": {}} for name in ("a", "b")],
+        }
+    )
+    context.skill_context.append("new hook")
+    await context.add_message({"role": "tool", "tool_call_id": "a", "content": "A"})
+    pending = await context.get_messages_for_request(token_budget=BUDGET)
+    assert wire(pending[: len(first)]) == wire(first)
+    assert "new hook" not in wire(pending)
+    await context.add_message({"role": "tool", "tool_call_id": "b", "content": "B"})
+    complete = await context.get_messages_for_request(token_budget=BUDGET)
+    assert wire(complete[: len(pending)]) == wire(pending)
+    assert [item["role"] for item in complete[-4:]] == ["assistant", "tool", "tool", "user"]
+    assert complete[-1]["content"] == "new hook"

tests/e2e/python/test_skill_lifecycle.py

@@ -526,3 +526,59 @@ async def test_malformed_skill_declarations_have_named_input_errors(monkeypatch,
         await create_agent(options(tmp_path))
     assert error.value.code == "invalid_input"
     assert cause in error.value.message
+
+
+@pytest.mark.parametrize("ending", ["success", "failure", "cancelled"])
+async def test_hook_batches_keep_request_prefix_and_expire_on_every_turn_exit(monkeypatch, tmp_path, ending):
+    skill = write_skill(
+        tmp_path,
+        hooks={
+            name: [{**command(f"printf 'context from {name}'"), "matcher": "counter"}]
+            for name in ("PreToolUse", "PostToolUse")
+        },
+    )
+    final = {"success": {"text": "Complete"}, "failure": {"failure": True}, "cancelled": {"block": True}}[ending]
+    factories = provision(
+        monkeypatch,
+        [
+            invoke("load_skill", name="review"),
+            invoke("counter"),
+            invoke("counter"),
+            final,
+            {"text": "Next turn"},
+        ],
+    )
+
+    async def counter(arguments, context):
+        return "counted"
+
+    tool = Tool("counter", "Count work.", {"$schema": SCHEMA, "type": "object"}, counter)
+    async with (
+        await create_agent(options(skill, tools=[*BUILTIN_TOOLS, tool])) as agent,
+        await agent.create_session(SessionOptions(persistence="ephemeral")) as session,
+    ):
+        turn = await start(session)
+
+        async def collect():
+            return [event async for event in turn.events()]
+
+        collecting = asyncio.create_task(collect())
+        if ending == "cancelled":
+            async with asyncio.timeout(5):
+                while not factories or len(factories[0].requests) < 4:
+                    await asyncio.sleep(0.01)
+            await turn.cancel()
+        stream = await asyncio.wait_for(collecting, 5)
+        assert stream[-1].payload.state == ending
+        paired(stream)
+        requests = factories[0].requests
+        before, after = requests[2]["messages"], requests[3]["messages"]
+        assert json.dumps(after[: len(before)], sort_keys=True) == json.dumps(before, sort_keys=True)
+        for name in ("PreToolUse", "PostToolUse"):
+            assert name in str(before[-1]["content"])
+            assert name in str(after[-1]["content"])
+        assert (await events(session))[-1].payload.state == "success"
+        assert not any(
+            message["role"] == "user" and "context from" in str(message["content"])
+            for message in requests[4]["messages"]
+        )