diff --git a/AGENTS.md b/AGENTS.md
index fc9a9704..bede34cb 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -100,6 +100,21 @@ dynamic temporary stdout capture.
Run-command diagnostics before headless execution belong on stderr so JSON
stdout remains one parseable payload.
+## Fork event ownership and cumulative cost
+
+Forks NEVER COPY parent root events, CI capture, or lifecycle metadata; parent
+CI is read only to build the boundary, and resume never reads ancestors.
+`--no-events` remains an accepted compatibility option only; event activity
+stays owned by its emitter.
+For a resumed transcript fork (identified by `forked_from_turn`, never
+`parent_id` alone), the child saves a versioned immutable
+`fork_cost_boundary`: inherited-turn cumulative Decimal totals, a canonical
+prefix fingerprint, and owner/fence provenance. Resume reads that verified
+boundary plus child-owned CI cost only; it never reconstructs ancestors.
+Honor configured CI relocation without falling back from a missing selected
+capture. An unavailable or pre-boundary fork is explicitly incomplete, never
+verified zero. Keep ordinary non-fork resume's CI-then-native fallback.
+
Tests must use the Foundation dependency installed in their test environment.
Do not prepend a neighboring checkout to `sys.path`: that silently bypasses the
published dependency and lockfile. Install an explicit local override in the
diff --git a/amplifier_app_cli/commands/session.py b/amplifier_app_cli/commands/session.py
index 5da789bb..b3ad551d 100644
--- a/amplifier_app_cli/commands/session.py
+++ b/amplifier_app_cli/commands/session.py
@@ -83,29 +83,6 @@ def _fork_source(store: SessionStore, session_id: str):
yield load_root_resume(store, session_id)
-def _copy_legacy_fork_events(parent_dir: Path, child_dir: Path, turn: int, child_id: str, parent_id: str) -> int:
- """Preserve the CLI's optional legacy log copy, never copy CI evidence.
-
- Context Intelligence captures stay owned by their original session. Fork
- lineage links to them; the child's logger records only subsequent activity.
- """
- from amplifier_foundation.session.events import slice_events_for_fork
-
- source = parent_dir / "events.jsonl"
- if not source.is_file():
- return 0
- destination = child_dir / "events.jsonl"
- try:
- return slice_events_for_fork(
- source, parent_dir / "transcript.jsonl", turn, destination,
- new_session_id=child_id, parent_session_id=parent_id,
- )
- except Exception:
- # Audit copying is best-effort and cannot change recovered messages.
- destination.write_text("", encoding="utf-8")
- return 0
-
-
def _record_bundle_override(
metadata: dict, new_bundle: str, original_config: str
) -> None:
@@ -885,7 +862,11 @@ def sessions_show(
is_flag=True,
help="Resume forked session immediately",
)
- @click.option("--no-events", is_flag=True, help="Skip copying the legacy root events.jsonl (CI activity stays with its parent)")
+ @click.option(
+ "--no-events",
+ is_flag=True,
+ help="Compatibility option; forks never copy event logs.",
+ )
def sessions_fork(
session_id: str,
turn: int | None,
@@ -1040,27 +1021,34 @@ def sessions_fork(
action_messages, turn=turn, parent_id=session_id
)
child_id = new_name or result.session_id
+ child_messages = result.messages or []
+ from ..cost_history import build_fork_cost_boundary
+
+ boundary = build_fork_cost_boundary(
+ parent_dir=session_dir,
+ parent_id=session_id,
+ parent_messages=action_messages,
+ parent_metadata=action_metadata,
+ fork_turn=result.forked_from_turn,
+ child_messages=child_messages,
+ )
now = datetime.now(UTC).isoformat()
store.save_new(
child_id,
- result.messages or [],
+ child_messages,
{
"session_id": child_id,
"parent_id": session_id,
"forked_from_turn": result.forked_from_turn,
+ "fork_cost_boundary": boundary,
"forked_at": now,
"created": now,
- "turn_count": count_turns(result.messages or []),
+ "turn_count": count_turns(child_messages),
"bundle": action_metadata.get("bundle"),
"model": action_metadata.get("model"),
},
)
result.session_id = child_id
- result.session_dir = store.base_dir / child_id
- if not no_events:
- result.events_count = _copy_legacy_fork_events(
- session_dir, result.session_dir, turn, child_id, session_id
- )
console.print(
f"[green]✓[/green] Forked session created: {result.session_id}"
@@ -1068,8 +1056,9 @@ def sessions_fork(
console.print(f" Messages: {result.message_count}")
console.print(f" Parent: {result.parent_id[:8]}...")
console.print(f" Forked at turn: {result.forked_from_turn}")
- if result.events_count > 0:
- console.print(f" Events copied: {result.events_count}")
+ console.print(" Event history remains with its original owners.")
+ if no_events:
+ console.print(" --no-events accepted for compatibility (no event logs are copied).")
console.print()
console.print(
f"Resume with: [cyan]amplifier session resume {result.session_id[:8]}[/cyan]"
diff --git a/amplifier_app_cli/cost_history.py b/amplifier_app_cli/cost_history.py
index c8d2a67a..a19346ab 100644
--- a/amplifier_app_cli/cost_history.py
+++ b/amplifier_app_cli/cost_history.py
@@ -1,73 +1,82 @@
-"""Restore cumulative session cost on resume (issue #284).
-
-Session LLM cost is accumulated **in-memory** in each provider module's
-``mount()`` closure (a ``_totals`` dict) and contributed to the kernel's
-``session.cost`` channel. When a session is resumed the provider re-mounts with
-that accumulator back at zero, so the cumulative session-cost counter restarts
-from zero. (Per-turn cost still displays correctly; only the running session
-total is lost.)
-
-This module reads the prior cumulative cost from the session's persisted
-``events.jsonl`` -- every ``llm:response`` event carries ``data.usage.cost_usd``
--- and re-seeds the running total by registering a synthetic ``session.cost``
-contributor on the *resumed session's own coordinator*. This mirrors the
-``register_contributor`` pattern already used by
-``amplifier_foundation.bridge_child_cost`` to bridge child-session cost into a
-parent.
-
-Design notes:
-- ``register_contributor`` APPENDS (the kernel never overwrites on duplicate
- name), and ``collect_contributions`` sums every registered contributor. The
- fresh per-mount provider accumulator starts at zero on resume and only counts
- turns executed *after* the resume, so the historical contributor and the live
- provider contributor never double-count the same spend.
-- Everything here is best-effort and must never break session startup: missing
- or corrupt event files simply yield no restored cost.
-"""
+"""Restore cumulative session cost without mutable fork ancestry."""
from __future__ import annotations
+import hashlib
+import json
import logging
import os
+from collections import Counter
+from dataclasses import dataclass
from decimal import Decimal
from decimal import InvalidOperation
from pathlib import Path
from typing import Any
+from amplifier_foundation import sanitize_message
+from amplifier_foundation.session import count_turns, get_turn_boundaries, slice_to_turn
from amplifier_foundation.session.history import SessionHistoryStore
+from amplifier_foundation.session.messages import is_real_user_message
logger = logging.getLogger(__name__)
-# Channel + event/field names defined by the kernel and provider modules.
SESSION_COST_CHANNEL = "session.cost"
_LLM_RESPONSE_EVENT = "llm:response"
+_PROMPT_SUBMIT_EVENT = "prompt:submit"
+_RESTORED_LINEAGE_MARKER = "amplifier_app_cli.restored_fork_cost"
+_BOUNDARY_KEY = "fork_cost_boundary"
+_BOUNDARY_VERSION = 1
+_MAX_BOUNDARY_TURNS = 4_096
+_MAX_BOUNDARY_BYTES = 64 * 1024
-def session_events_path(session_dir: Path) -> Path:
- """Resolve CI's reader-visible relocation before the old CLI log fallback.
+@dataclass(frozen=True)
+class ForkLineageCost:
+ """The restored fork subtotal and reasons that inherited evidence is incomplete."""
- Foundation deliberately does not choose process environment policy. The CI
- base path is a projects root; preserve the native project slug/session ID.
- Like CI, empty, unexpanded, or relative values are not relocation roots.
- """
+ total: Decimal
+ diagnostics: tuple[str, ...]
+
+ @property
+ def incomplete(self) -> bool:
+ return bool(self.diagnostics)
+
+
+@dataclass(frozen=True)
+class _CaptureCosts:
+ by_turn: dict[int, Decimal]
+ diagnostics: tuple[str, ...]
+ final_fence_line: int | None
+ warnings: tuple[str, ...] = ()
+
+
+def ci_events_path(session_dir: Path) -> Path:
+ """Return the only CI capture selected by the active relocation policy."""
raw_root = os.environ.get("AMPLIFIER_CONTEXT_INTELLIGENCE_BASE_PATH", "").strip()
root = Path(raw_root).expanduser() if raw_root and "${" not in raw_root else None
- events_path = SessionHistoryStore(session_dir).events_path
if root is not None and root.is_absolute():
project_slug = session_dir.parent.parent.name
capture_dir = root / project_slug / "sessions" / session_dir.name
- events_path = SessionHistoryStore(capture_dir).events_path
- # Never combine both captures: that can count the same model call twice.
+ return SessionHistoryStore(capture_dir).events_path
+ return SessionHistoryStore(session_dir).events_path
+
+
+def session_events_path(session_dir: Path) -> Path:
+ """Resolve CI relocation before the ordinary-session native-log fallback."""
+ events_path = ci_events_path(session_dir)
return events_path if events_path.exists() else session_dir / "events.jsonl"
-def sum_prior_cost_usd(events_path: Path, *, session_id: str | None = None) -> Decimal | None:
- """Read cost from the shared event reader without constructing a message cache.
+def _decimal(value: object) -> Decimal | None:
+ try:
+ amount = Decimal(str(value))
+ except (InvalidOperation, ValueError):
+ return None
+ return amount if amount.is_finite() and amount >= 0 else None
- The path is explicit so old CLI root event logs remain supported when the
- host chooses that fallback. CI records are normalized/scoped by Foundation.
- Malformed events and invalid costs never prevent session startup.
- """
+
+def sum_prior_cost_usd(events_path: Path, *, session_id: str | None = None) -> Decimal | None:
+ """Read ordinary-session cost through Foundation's streaming CI reader."""
session_dir = events_path.parent
if session_dir.name == "context-intelligence":
session_dir = session_dir.parent
@@ -77,42 +86,529 @@ def sum_prior_cost_usd(events_path: Path, *, session_id: str | None = None) -> D
if event.get("event") != _LLM_RESPONSE_EVENT:
continue
usage = event["data"].get("usage")
- cost = usage.get("cost_usd") if isinstance(usage, dict) else None
- if cost is None:
- continue
- try:
- amount = Decimal(str(cost))
- if not amount.is_finite() or amount < 0:
- continue
+ amount = _decimal(usage.get("cost_usd") if isinstance(usage, dict) else None)
+ if amount is not None:
total = (total or Decimal("0")) + amount
- except (InvalidOperation, ValueError):
- continue
return total
+def _add(diagnostics: set[str], code: str) -> None:
+ diagnostics.add(code)
+
+
+def _capture_stamp(path: Path) -> tuple[int, int, int] | None:
+ try:
+ stat = path.stat()
+ except OSError:
+ return None
+ return stat.st_ino, stat.st_size, stat.st_mtime_ns
+
+
+@dataclass(frozen=True)
+class _NativeTurn:
+ """One Foundation turn and its optional CI-prompt anchor."""
+
+ number: int
+ message_index: int
+ prompt: str | None
+ unsupported: bool
+
+
+def _result_ids(message: dict[str, Any]) -> set[str]:
+ """Match Foundation history's persisted tool-result identity rules."""
+ found: set[str] = set()
+ if isinstance(message.get("tool_call_id"), str):
+ found.add(message["tool_call_id"])
+ content = message.get("content")
+ if isinstance(content, list):
+ for block in content:
+ if not isinstance(block, dict) or block.get("type") not in (
+ "tool_result",
+ "function_call_output",
+ ):
+ continue
+ identity = (
+ block.get("tool_use_id")
+ or block.get("tool_call_id")
+ or block.get("call_id")
+ or block.get("id")
+ )
+ if isinstance(identity, str):
+ found.add(identity)
+ return found
+
+
+def _prompt_text(message: dict[str, Any]) -> str | None:
+ """Extract exactly the prompt text emitted by supported live prompt builders."""
+ content = message.get("content")
+ if isinstance(content, str):
+ return content
+ if isinstance(content, list):
+ for block in content:
+ if isinstance(block, dict) and block.get("type") == "text":
+ text = block.get("text")
+ if isinstance(text, str):
+ return text
+ return None
+
+
+def _native_turns(messages: list[dict[str, Any]]) -> list[_NativeTurn]:
+ """Project every Foundation user turn onto optional real-human prompt anchors."""
+ turns: list[_NativeTurn] = []
+ for number, message_index in enumerate(get_turn_boundaries(messages), start=1):
+ message = messages[message_index]
+ metadata = message.get("metadata") if isinstance(message.get("metadata"), dict) else {}
+ is_anchor = (
+ is_real_user_message(message)
+ and not metadata.get("ephemeral")
+ and not _result_ids(message)
+ )
+ prompt = _prompt_text(message) if is_anchor else None
+ turns.append(_NativeTurn(number, message_index, prompt, is_anchor and prompt is None))
+ return turns
+
+
+def _canonical_saved_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]] | None:
+ """Mirror SessionStore's transcript persistence before hashing a prefix."""
+ canonical: list[dict[str, Any]] = []
+ try:
+ for message in messages:
+ if not isinstance(message, dict):
+ return None
+ if message.get("role") in ("system", "developer"):
+ continue
+ canonical.append(sanitize_message(message))
+ except Exception:
+ return None
+ return canonical
+
+
+def _prefix_fingerprint(messages: list[dict[str, Any]], turns: int) -> str | None:
+ """Hash Foundation's canonical native-turn prefix without retaining it."""
+ canonical = _canonical_saved_messages(messages)
+ if canonical is None or turns < 0 or turns > count_turns(canonical):
+ return None
+ try:
+ prefix = (
+ []
+ if turns == 0
+ else slice_to_turn(canonical, turns, handle_orphaned_tools="complete")
+ )
+ encoded = json.dumps(
+ prefix, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False
+ ).encode("utf-8")
+ except (TypeError, ValueError):
+ return None
+ return hashlib.sha256(encoded).hexdigest()
+
+
+def _history_diagnostics(history: SessionHistoryStore, owner: str) -> set[str]:
+ return {f"ci_{item.code}:{owner}" for item in history.diagnostics}
+
+
+def _foreign_owner_cost_exists(events_path: Path, owner: str) -> bool:
+ """Report foreign LLM cost rows Foundation intentionally filters out."""
+ try:
+ stream = events_path.open("rb")
+ except OSError:
+ return False
+ with stream:
+ for raw in stream:
+ try:
+ record = json.loads(raw)
+ except (TypeError, UnicodeError, json.JSONDecodeError):
+ continue
+ if not isinstance(record, dict) or record.get("event") != _LLM_RESPONSE_EVENT:
+ continue
+ data = record.get("data")
+ if not isinstance(data, dict):
+ continue
+ identity = data.get("session_id") or record.get("session_id")
+ if isinstance(identity, str) and identity and identity != owner:
+ usage = data.get("usage")
+ if isinstance(usage, dict) and "cost_usd" in usage:
+ return True
+ return False
+
+
+def _scan_owner_turn_costs(
+ session_dir: Path,
+ owner: str,
+ messages: list[dict[str, Any]],
+ required_turns: range,
+ *,
+ inherited_native_turn_count: int,
+) -> _CaptureCosts:
+ """Associate owner costs with real prompt anchors in native turn space.
+
+ Foundation defines every ``role=user`` row as a turn. Only non-ephemeral,
+ non-tool-result human rows can be CI prompt anchors. Injected rows consume a
+ native turn but carry the preceding anchored cost rather than inventing a
+ submit, response, or fence of their own.
+ """
+ events_path = ci_events_path(session_dir)
+ diagnostics: set[str] = set()
+ native = _native_turns(messages)
+ required = set(required_turns)
+ if not required:
+ return _CaptureCosts({}, (), None)
+ if max(required) > len(native) or inherited_native_turn_count < 0:
+ return _CaptureCosts({}, (f"invalid_fork_turn:{owner}",), None)
+
+ native_by_number = {turn.number: turn for turn in native}
+ warnings: set[str] = set()
+ for turn in required:
+ item = native_by_number[turn]
+ if item.unsupported:
+ _add(diagnostics, f"unsupported_prompt_anchor:{owner}:{turn}")
+ if item.prompt is None and _non_anchor_has_assistant_output(messages, item, native):
+ warnings.add(f"unmapped_non_anchor_output:{owner}:{turn}")
+ if diagnostics:
+ return _CaptureCosts({}, tuple(sorted(diagnostics)), None, tuple(sorted(warnings)))
+ if not events_path.is_file():
+ return _CaptureCosts({}, (f"missing_ci_capture:{owner}",), None, tuple(sorted(warnings)))
+
+ owner_anchors = [
+ turn for turn in native if turn.number > inherited_native_turn_count and turn.prompt is not None
+ ]
+ expected = [turn.prompt for turn in owner_anchors]
+ target_anchors = [
+ turn for turn in owner_anchors if turn.number in required
+ ]
+ before = _capture_stamp(events_path)
+ event_prompts: list[str] = []
+ first = SessionHistoryStore(session_dir, events_path=events_path, session_id=owner)
+ for event in first.iter_events():
+ if event.get("session_id") != owner or event.get("event") != _PROMPT_SUBMIT_EVENT:
+ continue
+ data = event.get("data")
+ prompt = data.get("prompt") if isinstance(data, dict) else None
+ if not isinstance(prompt, str):
+ _add(diagnostics, f"invalid_prompt_submit:{owner}")
+ continue
+ event_prompts.append(prompt)
+ diagnostics.update(_history_diagnostics(first, owner))
+ exact_sequence = not diagnostics and event_prompts == expected
+ transcript_counts = Counter(expected)
+ event_counts = Counter(event_prompts)
+ unique_turn_by_prompt = {
+ turn.prompt: turn.number
+ for turn in target_anchors
+ if transcript_counts[turn.prompt] == 1 and event_counts[turn.prompt] == 1
+ }
+
+ costs = {turn: Decimal("0") for turn in required}
+ submits: set[int] = set()
+ responses: set[int] = set()
+ current_turn: int | None = None
+ sequence_index = 0
+ final_fence_line: int | None = None
+ second = SessionHistoryStore(session_dir, events_path=events_path, session_id=owner)
+ for event in second.iter_events():
+ if event.get("session_id") != owner:
+ continue
+ if event.get("event") == _PROMPT_SUBMIT_EVENT:
+ data = event.get("data")
+ prompt = data.get("prompt") if isinstance(data, dict) else None
+ if exact_sequence and sequence_index < len(owner_anchors):
+ current_turn = owner_anchors[sequence_index].number
+ else:
+ current_turn = unique_turn_by_prompt.get(prompt) if isinstance(prompt, str) else None
+ sequence_index += 1
+ if current_turn in required:
+ submits.add(current_turn)
+ continue
+ if event.get("event") != _LLM_RESPONSE_EVENT or current_turn not in required:
+ continue
+ data = event.get("data")
+ usage = data.get("usage") if isinstance(data, dict) else None
+ if not isinstance(usage, dict) or "cost_usd" not in usage:
+ _add(diagnostics, f"missing_cost:{owner}")
+ continue
+ amount = _decimal(usage["cost_usd"])
+ if amount is None:
+ _add(diagnostics, f"invalid_cost:{owner}")
+ continue
+ costs[current_turn] += amount
+ responses.add(current_turn)
+ final_fence_line = event.get("line") if isinstance(event.get("line"), int) else final_fence_line
+ diagnostics.update(_history_diagnostics(second, owner))
+ if _foreign_owner_cost_exists(events_path, owner):
+ _add(diagnostics, f"foreign_cost:{owner}")
+ if before != _capture_stamp(events_path):
+ _add(diagnostics, f"unstable_ci_capture:{owner}")
+ for turn in required:
+ if native_by_number[turn].prompt is None:
+ continue
+ if turn not in submits:
+ _add(diagnostics, f"missing_prompt_fence:{owner}:{turn}")
+ elif turn not in responses:
+ _add(diagnostics, f"missing_response_cost:{owner}:{turn}")
+ return _CaptureCosts(costs, tuple(sorted(diagnostics)), final_fence_line, tuple(sorted(warnings)))
+
+
+def _non_anchor_has_assistant_output(
+ messages: list[dict[str, Any]], turn: _NativeTurn, native: list[_NativeTurn]
+) -> bool:
+ """Identify output after an injected turn that cannot prove a CI mapping."""
+ next_index = next(
+ (item.message_index for item in native if item.number == turn.number + 1),
+ len(messages),
+ )
+ return any(message.get("role") == "assistant" for message in messages[turn.message_index + 1 : next_index])
+
+
+def _has_unprovable_intra_prompt_cutoff(
+ messages: list[dict[str, Any]], native: list[_NativeTurn], fork_turn: int
+) -> bool:
+ """Reject a cut that excludes injected output before the next real prompt."""
+ for turn in native:
+ if turn.number <= fork_turn:
+ continue
+ if turn.prompt is not None or turn.unsupported:
+ return False
+ if _non_anchor_has_assistant_output(messages, turn, native):
+ return True
+ return False
+
+
+def _unavailable_boundary(owner: str, *reasons: str) -> dict[str, Any]:
+ return {
+ "version": _BOUNDARY_VERSION,
+ "status": "unavailable",
+ "inherited_turn_count": 0,
+ "cumulative_cost_usd_by_turn": [],
+ "prefix_fingerprint": None,
+ "provenance": {"owner": owner, "fence": "unavailable"},
+ "reasons": sorted(set(reasons)) or ["unavailable"],
+ }
+
+
+def _verified_boundary(metadata: dict[str, Any]) -> tuple[list[Decimal] | None, int, tuple[str, ...]]:
+ """Validate the persisted schema of an immutable cost projection."""
+ raw = metadata.get(_BOUNDARY_KEY)
+ if not isinstance(raw, dict):
+ return None, 0, ("missing_fork_cost_boundary",)
+ if raw.get("version") != _BOUNDARY_VERSION:
+ return None, 0, ("unsupported_fork_cost_boundary",)
+ try:
+ encoded_size = len(json.dumps(raw, ensure_ascii=False, allow_nan=False).encode("utf-8"))
+ except (TypeError, ValueError):
+ return None, 0, ("invalid_fork_cost_boundary",)
+ if encoded_size > _MAX_BOUNDARY_BYTES:
+ return None, 0, ("oversized_fork_cost_boundary",)
+ if raw.get("status") != "verified":
+ reasons = raw.get("reasons")
+ stable = tuple(str(item) for item in reasons) if isinstance(reasons, list) else ()
+ return None, 0, ("unavailable_fork_cost_boundary", *stable)
+ turns = raw.get("inherited_turn_count")
+ values = raw.get("cumulative_cost_usd_by_turn")
+ fingerprint = raw.get("prefix_fingerprint")
+ if (
+ type(turns) is not int
+ or turns < 0
+ or turns > _MAX_BOUNDARY_TURNS
+ or not isinstance(values, list)
+ or len(values) != turns
+ or not isinstance(fingerprint, str)
+ or len(fingerprint) != 64
+ ):
+ return None, 0, ("invalid_fork_cost_boundary",)
+ totals: list[Decimal] = []
+ prior = Decimal("0")
+ for value in values:
+ amount = _decimal(value)
+ if amount is None or amount < prior:
+ return None, 0, ("invalid_fork_cost_boundary",)
+ totals.append(amount)
+ prior = amount
+ return totals, turns, ()
+
+
+def _verified_boundary_prefix(
+ metadata: dict[str, Any], messages: list[dict[str, Any]]
+) -> tuple[list[Decimal] | None, int, tuple[str, ...]]:
+ """Validate a boundary before projecting it into a new fork."""
+ totals, turns, errors = _verified_boundary(metadata)
+ if errors or totals is None:
+ return totals, turns, errors
+ raw = metadata[_BOUNDARY_KEY]
+ if _prefix_fingerprint(messages, turns) != raw["prefix_fingerprint"]:
+ return None, 0, ("fork_prefix_changed",)
+ return totals, turns, ()
+
+
+def build_fork_cost_boundary(
+ *,
+ parent_dir: Path,
+ parent_id: str,
+ parent_messages: list[dict[str, Any]],
+ parent_metadata: dict[str, Any],
+ fork_turn: int,
+ child_messages: list[dict[str, Any]],
+) -> dict[str, Any]:
+ """Build the immutable inherited-cost projection written with a child."""
+ canonical_parent = _canonical_saved_messages(parent_messages)
+ canonical_child = _canonical_saved_messages(child_messages)
+ if canonical_parent is None or canonical_child is None:
+ return _unavailable_boundary(parent_id, "invalid_inherited_prefix")
+ child_turns = count_turns(canonical_child)
+ if fork_turn < 1 or fork_turn != child_turns:
+ return _unavailable_boundary(parent_id, "invalid_fork_turn")
+ if fork_turn > _MAX_BOUNDARY_TURNS:
+ return _unavailable_boundary(parent_id, "fork_cost_boundary_too_large")
+ if _has_unprovable_intra_prompt_cutoff(
+ canonical_parent, _native_turns(canonical_parent), fork_turn
+ ):
+ return _unavailable_boundary(parent_id, "unprovable_intra_prompt_cutoff")
+
+ inherited: list[Decimal] = []
+ parent_inherited_turns = 0
+ warnings: tuple[str, ...] = ()
+ if "forked_from_turn" in parent_metadata:
+ inherited, parent_inherited_turns, errors = _verified_boundary_prefix(
+ parent_metadata, canonical_parent
+ )
+ if errors or inherited is None:
+ return _unavailable_boundary(parent_id, *errors)
+
+ if fork_turn <= parent_inherited_turns:
+ totals = inherited[:fork_turn]
+ fence: dict[str, Any] = {"kind": "projected_boundary", "turn": fork_turn}
+ else:
+ observation = _scan_owner_turn_costs(
+ parent_dir,
+ parent_id,
+ canonical_parent,
+ range(parent_inherited_turns + 1, fork_turn + 1),
+ inherited_native_turn_count=parent_inherited_turns,
+ )
+ if observation.diagnostics:
+ return _unavailable_boundary(parent_id, *observation.diagnostics)
+ warnings = observation.warnings
+ totals = list(inherited)
+ running = totals[-1] if totals else Decimal("0")
+ for turn in range(parent_inherited_turns + 1, fork_turn + 1):
+ running += observation.by_turn[turn]
+ totals.append(running)
+ fence = {
+ "kind": "capture_end",
+ "line": observation.final_fence_line,
+ "turn": fork_turn,
+ }
+
+ fingerprint = _prefix_fingerprint(canonical_child, fork_turn)
+ if fingerprint is None:
+ return _unavailable_boundary(parent_id, "invalid_inherited_prefix")
+ boundary: dict[str, Any] = {
+ "version": _BOUNDARY_VERSION,
+ "status": "verified",
+ "inherited_turn_count": fork_turn,
+ "cumulative_cost_usd_by_turn": [str(value) for value in totals],
+ "prefix_fingerprint": fingerprint,
+ "provenance": {"owner": parent_id, "fence": fence},
+ **({"warnings": list(warnings)} if warnings else {}),
+ }
+ if len(json.dumps(boundary, ensure_ascii=False).encode("utf-8")) > _MAX_BOUNDARY_BYTES:
+ return _unavailable_boundary(parent_id, "fork_cost_boundary_too_large")
+ return boundary
+
+
+def _sum_owner_ci_cost(session_dir: Path, owner: str) -> tuple[Decimal, tuple[str, ...]]:
+ """Sum child-owned spend only; inherited cost comes only from metadata."""
+ events_path = ci_events_path(session_dir)
+ diagnostics: set[str] = set()
+ if not events_path.is_file():
+ return Decimal("0"), (f"missing_ci_capture:{owner}",)
+ before = _capture_stamp(events_path)
+ history = SessionHistoryStore(session_dir, events_path=events_path, session_id=owner)
+ total = Decimal("0")
+ for event in history.iter_events():
+ if event.get("session_id") != owner:
+ continue
+ if event.get("event") != _LLM_RESPONSE_EVENT:
+ continue
+ usage = event["data"].get("usage")
+ if not isinstance(usage, dict) or "cost_usd" not in usage:
+ _add(diagnostics, f"missing_cost:{owner}")
+ continue
+ amount = _decimal(usage["cost_usd"])
+ if amount is None:
+ _add(diagnostics, f"invalid_cost:{owner}")
+ continue
+ total += amount
+ diagnostics.update(_history_diagnostics(history, owner))
+ if _foreign_owner_cost_exists(events_path, owner):
+ _add(diagnostics, f"foreign_cost:{owner}")
+ if before != _capture_stamp(events_path):
+ _add(diagnostics, f"unstable_ci_capture:{owner}")
+ return total, tuple(sorted(diagnostics))
+
+
+def _restoration_marker(coordinator: Any) -> set[str] | None:
+ state = getattr(coordinator, "session_state", None)
+ if not isinstance(state, dict):
+ return None
+ marker = state.get(_RESTORED_LINEAGE_MARKER)
+ if marker is None:
+ marker = set()
+ state[_RESTORED_LINEAGE_MARKER] = marker
+ return marker if isinstance(marker, set) else None
+
+
+def restore_fork_lineage_cost(
+ coordinator: Any, *, session_id: str, session_dir: Path
+) -> ForkLineageCost:
+ """Restore immutable inherited cost plus current child-owned CI spend.
+
+ No parent directory is resolved or read. Pre-boundary forks are incomplete,
+ not reconstructed from mutable timestamps or native event logs.
+ """
+ diagnostics: set[str] = set()
+ try:
+ history = SessionHistoryStore(session_dir)
+ metadata = history.load_metadata()
+ except Exception:
+ metadata = {}
+ _add(diagnostics, "unreadable_fork_metadata")
+ inherited, _, errors = _verified_boundary(metadata)
+ diagnostics.update(errors)
+ inherited_total = inherited[-1] if inherited else Decimal("0")
+ own_total, own_diagnostics = _sum_owner_ci_cost(session_dir, session_id)
+ diagnostics.update(own_diagnostics)
+ total = inherited_total + own_total
+
+ marker = _restoration_marker(coordinator)
+ if total > 0 and (marker is None or session_id not in marker):
+ try:
+ coordinator.register_contributor(
+ SESSION_COST_CHANNEL,
+ f"history:{session_id}",
+ lambda total=total: {"cost_usd": str(total)},
+ )
+ except Exception:
+ logger.warning(
+ "Failed to restore fork lineage cost for %s; continuing without it",
+ session_id,
+ exc_info=True,
+ )
+ _add(diagnostics, f"registration_failed:{session_id}")
+ else:
+ if marker is not None:
+ marker.add(session_id)
+ return ForkLineageCost(total=total, diagnostics=tuple(sorted(diagnostics)))
+
+
def restore_session_cost(
coordinator: Any,
session_id: str,
events_path: Path,
) -> Decimal | None:
- """Re-seed cumulative session cost on resume via a synthetic contributor.
-
- Reads the prior cumulative cost from ``events_path`` and, when cost data
- exists, registers a ``session.cost`` contributor on ``coordinator`` so that
- ``collect_contributions("session.cost")`` reports the pre-resume total
- alongside the fresh per-mount provider contributions.
-
- Returns the restored total (a ``Decimal``), or ``None`` when there was no
- prior cost to restore or registration failed. Never raises.
- """
+ """Re-seed ordinary resume cost via a synthetic contributor."""
prior_total = sum_prior_cost_usd(events_path, session_id=session_id)
if prior_total is None or prior_total <= 0:
return None
-
try:
- # Freeze the total into the callback default so it is captured by value,
- # and stringify to match the provider modules' contributor payloads
- # (Decimal is not JSON-serializable; sum_cost_usd accepts str or Decimal).
coordinator.register_contributor(
SESSION_COST_CHANNEL,
f"history:{session_id}",
@@ -125,10 +621,5 @@ def restore_session_cost(
exc_info=True,
)
return None
-
- logger.info(
- "Restored prior session cost $%s for resumed session %s",
- prior_total,
- session_id,
- )
- return prior_total
+ logger.info("Restored $%s cumulative session cost for %s", prior_total, session_id)
+ return prior_total
\ No newline at end of file
diff --git a/amplifier_app_cli/main.py b/amplifier_app_cli/main.py
index 140233bc..1598711b 100644
--- a/amplifier_app_cli/main.py
+++ b/amplifier_app_cli/main.py
@@ -1901,7 +1901,6 @@ async def _fork_session(self, args: str) -> str:
try:
from amplifier_foundation.session import (
count_turns,
- fork_session,
fork_session_in_memory,
get_turn_summary,
)
@@ -2002,21 +2001,34 @@ async def _fork_session(self, args: str) -> str:
child_id = custom_name or result.session_id
now = datetime.now(UTC).isoformat()
shared_parent = root_state.read(store)
+ parent_messages = messages
parent_metadata = (
shared_parent[1]
if shared_parent is not None
else store.get_metadata_if_exists(session_id)
)
+ child_messages = result.messages or []
+ from .cost_history import build_fork_cost_boundary
+
+ boundary = build_fork_cost_boundary(
+ parent_dir=session_dir,
+ parent_id=session_id,
+ parent_messages=parent_messages,
+ parent_metadata=parent_metadata,
+ fork_turn=result.forked_from_turn,
+ child_messages=child_messages,
+ )
store.save_new(
child_id,
- result.messages or [],
+ child_messages,
{
"session_id": child_id,
"parent_id": session_id,
"forked_from_turn": result.forked_from_turn,
+ "fork_cost_boundary": boundary,
"forked_at": now,
"created": now,
- "turn_count": count_turns(result.messages or []),
+ "turn_count": count_turns(child_messages),
"bundle": parent_metadata.get("bundle", self.bundle_name),
"model": parent_metadata.get("model"),
},
@@ -2024,21 +2036,46 @@ async def _fork_session(self, args: str) -> str:
result.session_id = child_id
else:
child_id = custom_name or str(uuid.uuid4())
- store.reserve_session(child_id)
- result = fork_session(
- session_dir,
- turn=turn,
- new_session_id=child_id,
- include_events=True,
+ from .cost_history import build_fork_cost_boundary
+ from .shared_root_state import load_root_resume
+
+ parent_messages, parent_metadata = load_root_resume(store, session_id)
+ result = fork_session_in_memory(
+ parent_messages, turn=turn, parent_id=session_id
+ )
+ child_messages = result.messages or []
+ boundary = build_fork_cost_boundary(
+ parent_dir=session_dir,
+ parent_id=session_id,
+ parent_messages=parent_messages,
+ parent_metadata=parent_metadata,
+ fork_turn=result.forked_from_turn,
+ child_messages=child_messages,
+ )
+ now = datetime.now(UTC).isoformat()
+ store.save_new(
+ child_id,
+ child_messages,
+ {
+ "session_id": child_id,
+ "parent_id": session_id,
+ "forked_from_turn": result.forked_from_turn,
+ "fork_cost_boundary": boundary,
+ "forked_at": now,
+ "created": now,
+ "turn_count": count_turns(child_messages),
+ "bundle": parent_metadata.get("bundle", self.bundle_name),
+ "model": parent_metadata.get("model"),
+ },
)
+ result.session_id = child_id
lines = [
f"✓ Forked session created: {result.session_id}",
f" Messages: {result.message_count}",
f" Forked at turn: {result.forked_from_turn} of {max_turns}",
]
- if result.events_count > 0:
- lines.append(f" Events copied: {result.events_count}")
+ lines.append(" Event history remains with its original owners.")
lines.append("")
lines.append(
f"Resume with: amplifier session resume {result.session_id[:8]}"
diff --git a/amplifier_app_cli/session_runner.py b/amplifier_app_cli/session_runner.py
index c6342800..1f0a3df4 100644
--- a/amplifier_app_cli/session_runner.py
+++ b/amplifier_app_cli/session_runner.py
@@ -364,19 +364,44 @@ async def create_initialized_session(
"Context module lacks set_messages - transcript NOT restored"
)
- # Step 7.5: Restore cumulative session cost (resume only) - issue #284
- # Provider cost accumulators live in each provider's mount() closure and are
- # zeroed on resume, so the running session-cost total would otherwise restart
- # from zero. Re-seed the "session.cost" channel from the persisted
- # Context Intelligence events (or the old CLI log when absent) by
- # registering a synthetic historical contributor on this session's own
- # coordinator. Best-effort: never blocks startup.
+ # Step 7.5: Restore cumulative session cost (resume only) - issue #284.
+ # Forks must be identified by their native fork turn, not parent_id alone:
+ # agent children have parents too, but never inherit root conversation cost.
+ # A fork's native root log may contain rewritten inherited activity, so it
+ # restores CI segments from their original owners only; ordinary resumes
+ # retain the old CI-then-native fallback for compatibility.
if config.is_resume:
- from .cost_history import restore_session_cost, session_events_path
+ from .cost_history import (
+ restore_fork_lineage_cost,
+ restore_session_cost,
+ session_events_path,
+ )
try:
session_dir = SessionStore().base_dir / session_id
- restore_session_cost(session.coordinator, session_id, session_events_path(session_dir))
+ from amplifier_foundation.session.history import SessionHistoryStore
+
+ metadata = SessionHistoryStore(session_dir).load_metadata()
+ if "forked_from_turn" in metadata:
+ result = restore_fork_lineage_cost(
+ session.coordinator,
+ session_id=session_id,
+ session_dir=session_dir,
+ )
+ if result.incomplete:
+ console.print(
+ "[yellow]Warning:[/yellow] cumulative fork cost history is incomplete; "
+ "only verified available CI cost segments were restored."
+ )
+ logger.warning(
+ "Incomplete fork cost history for %s: %s",
+ session_id,
+ ", ".join(result.diagnostics),
+ )
+ else:
+ restore_session_cost(
+ session.coordinator, session_id, session_events_path(session_dir)
+ )
except Exception:
logger.debug("Prior session cost restore skipped", exc_info=True)
diff --git a/docs/warm-shared-root-state.md b/docs/warm-shared-root-state.md
index 2367f506..5cf92cc1 100644
--- a/docs/warm-shared-root-state.md
+++ b/docs/warm-shared-root-state.md
@@ -43,9 +43,15 @@ checkpoint sessions, reporting the reason rather than leaving stale legacy data
that could resurrect a deleted session.
-`session fork --no-events` controls only the older root `events.jsonl` copy.
-Context Intelligence captures are not copied or assigned a new owner during a
-fork: parent lineage identifies the original activity and the child logger
-captures new work. Cost restoration prefers the CI capture, honors the
-`AMPLIFIER_CONTEXT_INTELLIGENCE_BASE_PATH` projects-root relocation, and falls
-back to the legacy root log only when that CI capture is absent.
+Forks never copy a root `events.jsonl` or a Context Intelligence capture.
+`session fork --no-events` remains accepted only for command-line compatibility;
+it has no effect because native transcript lineage is always preserved and event
+activity remains with its original owner. At fork creation, the child metadata
+stores a compact, versioned `fork_cost_boundary`: cumulative inherited-turn
+costs, a canonical inherited-prefix fingerprint, and owner/fence provenance.
+Resume verifies that local immutable projection and adds only the child-owned
+CI capture; it never reads an ancestor. It honors
+`AMPLIFIER_CONTEXT_INTELLIGENCE_BASE_PATH` relocation and reports unavailable
+or pre-boundary history rather than falling back to a native root log, which
+can contain rewritten inherited activity. Ordinary non-fork resume retains its
+CI-then-legacy-root compatibility fallback.
diff --git a/tests/test_cost_history.py b/tests/test_cost_history.py
index d0d66f99..c32d5c44 100644
--- a/tests/test_cost_history.py
+++ b/tests/test_cost_history.py
@@ -1,213 +1,817 @@
-"""Tests for cumulative-session-cost restoration on resume (issue #284).
+"""Regression tests for ordinary and immutable fork cumulative cost recovery."""
-Covers amplifier_app_cli.cost_history:
-- sum_prior_cost_usd(): parse events.jsonl and sum data.usage.cost_usd
-- restore_session_cost(): register a synthetic session.cost contributor
-
-The persisted event shape mirrors a real llm:response line: cost lives at
-data.usage.cost_usd and is stringified (Decimal is not JSON-serializable).
-"""
+from __future__ import annotations
+import hashlib
import json
+import shutil
from decimal import Decimal
+from pathlib import Path
from unittest.mock import MagicMock
-from amplifier_app_cli.cost_history import restore_session_cost
-from amplifier_app_cli.cost_history import sum_prior_cost_usd
+from amplifier_app_cli.cost_history import (
+ _prefix_fingerprint,
+ build_fork_cost_boundary,
+ restore_fork_lineage_cost,
+ restore_session_cost,
+ sum_prior_cost_usd,
+)
+from amplifier_app_cli.session_store import SessionStore
+from amplifier_foundation import sanitize_message
+from amplifier_foundation.session import slice_to_turn
-def _write_events(path, events):
- """Write a list of event dicts as JSONL to path."""
- path.write_text("\n".join(json.dumps(e) for e in events) + "\n", encoding="utf-8")
+def _write_events(path: Path, events: list[dict]) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text("\n".join(json.dumps(event) for event in events) + "\n", encoding="utf-8")
-def _llm_response(cost, **usage):
- """Build an llm:response event with cost at data.usage.cost_usd."""
- usage = {"input_tokens": 10, "output_tokens": 5, **usage}
- if cost is not None:
- usage["cost_usd"] = cost
- return {"event": "llm:response", "data": {"model": "test", "usage": usage}}
+def _submit(owner: str, prompt: str) -> dict:
+ return {"event": "prompt:submit", "data": {"session_id": owner, "prompt": prompt}}
-# --------------------------------------------------------------------------
-# sum_prior_cost_usd
-# --------------------------------------------------------------------------
+def _response(owner: str, cost: str) -> dict:
+ return {
+ "event": "llm:response",
+ "data": {"session_id": owner, "usage": {"cost_usd": cost}},
+ }
-def test_sum_returns_none_for_missing_file(tmp_path):
- assert sum_prior_cost_usd(tmp_path / "nope.jsonl") is None
+def _messages(*prompts: str) -> list[dict]:
+ result = []
+ for prompt in prompts:
+ result.extend((
+ {"role": "user", "content": prompt},
+ {"role": "assistant", "content": f"{prompt} answer"},
+ ))
+ return result
-def test_sum_returns_none_when_no_cost_events(tmp_path):
- events = tmp_path / "events.jsonl"
- _write_events(events, [{"event": "session:start", "data": {}}])
- assert sum_prior_cost_usd(events) is None
+def _registered(coordinator: MagicMock) -> dict:
+ captured = {}
+ coordinator.register_contributor = lambda channel, name, callback: captured.setdefault(
+ (channel, name), callback
+ )
+ return captured
+
+
+def _save(store: SessionStore, session_id: str, messages: list[dict], metadata: dict) -> Path:
+ store.save_new(session_id, messages, {"session_id": session_id, **metadata})
+ return store.base_dir / session_id
+
+
+def _fork(
+ store: SessionStore,
+ parent_id: str,
+ child_id: str,
+ parent_messages: list[dict],
+ parent_metadata: dict,
+ turn: int,
+) -> tuple[Path, dict]:
+ child_messages = parent_messages[: turn * 2]
+ boundary = build_fork_cost_boundary(
+ parent_dir=store.base_dir / parent_id,
+ parent_id=parent_id,
+ parent_messages=parent_messages,
+ parent_metadata={"session_id": parent_id, **parent_metadata},
+ fork_turn=turn,
+ child_messages=child_messages,
+ )
+ child = _save(
+ store,
+ child_id,
+ child_messages,
+ {
+ "parent_id": parent_id,
+ "forked_from_turn": turn,
+ "fork_cost_boundary": boundary,
+ },
+ )
+ return child, boundary
-def test_sum_single_llm_response(tmp_path):
+def test_ordinary_resume_keeps_legacy_cost_reader(tmp_path: Path) -> None:
events = tmp_path / "events.jsonl"
- _write_events(events, [_llm_response("0.178059")])
- assert sum_prior_cost_usd(events) == Decimal("0.178059")
+ _write_events(events, [_response("root", "0.10"), _response("root", "0.05")])
+ coordinator = MagicMock()
+ callbacks = _registered(coordinator)
+ assert sum_prior_cost_usd(events, session_id="root") == Decimal("0.15")
+ assert restore_session_cost(coordinator, "root", events) == Decimal("0.15")
+ assert callbacks[("session.cost", "history:root")]() == {"cost_usd": "0.15"}
-def test_sum_multiple_llm_responses(tmp_path):
- events = tmp_path / "events.jsonl"
+
+def test_timestamp_free_store_messages_create_verified_boundary_and_restore(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ parent_messages = _messages("first")
+ parent = _save(store, "parent", parent_messages, {})
_write_events(
- events,
- [
- {"event": "session:start", "data": {}},
- _llm_response("0.10"),
- {"event": "tool:call", "data": {}},
- _llm_response("0.05"),
- _llm_response("0.02"),
- ],
+ parent / "context-intelligence" / "events.jsonl",
+ [_submit("parent", "first"), _response("parent", "0.10")],
)
- assert sum_prior_cost_usd(events) == Decimal("0.17")
+ child, boundary = _fork(store, "parent", "child", parent_messages, {}, 1)
+ _write_events(child / "context-intelligence" / "events.jsonl", [_response("child", "0.20")])
+ coordinator = MagicMock(session_state={})
+ callbacks = _registered(coordinator)
-def test_sum_ignores_null_cost_and_bad_lines(tmp_path):
- events = tmp_path / "events.jsonl"
- # A malformed JSON line, a null-cost response, and a valid one.
- content = "\n".join(
- [
- "{not valid json",
- json.dumps(_llm_response(None)),
- json.dumps(_llm_response("0.03")),
- ]
+ result = restore_fork_lineage_cost(coordinator, session_id="child", session_dir=child)
+
+ assert all("timestamp" not in row for row in parent_messages)
+ assert boundary["status"] == "verified"
+ assert boundary["cumulative_cost_usd_by_turn"] == ["0.10"]
+ assert result == type(result)(Decimal("0.30"), ())
+ assert callbacks[("session.cost", "history:child")]() == {"cost_usd": "0.30"}
+
+
+def test_boundary_fingerprint_matches_session_store_sanitization(tmp_path: Path) -> None:
+ class NonSerializable:
+ pass
+
+ store = SessionStore(tmp_path / "sessions")
+ parent_messages = _messages("first")
+ parent = _save(store, "parent", parent_messages, {})
+ _write_events(
+ parent / "context-intelligence" / "events.jsonl",
+ [_submit("parent", "first"), _response("parent", "0.10")],
+ )
+ child_messages = _messages("first")
+ child_messages[1]["thinking_block"] = NonSerializable()
+ boundary = build_fork_cost_boundary(
+ parent_dir=parent,
+ parent_id="parent",
+ parent_messages=parent_messages,
+ parent_metadata={},
+ fork_turn=1,
+ child_messages=child_messages,
+ )
+ child = _save(
+ store,
+ "child",
+ child_messages,
+ {
+ "parent_id": "parent",
+ "forked_from_turn": 1,
+ "fork_cost_boundary": boundary,
+ },
)
- events.write_text(content + "\n", encoding="utf-8")
- assert sum_prior_cost_usd(events) == Decimal("0.03")
+ result = restore_fork_lineage_cost(MagicMock(session_state={}), session_id="child", session_dir=child)
-def test_sum_ignores_cost_on_non_llm_response_events(tmp_path):
- events = tmp_path / "events.jsonl"
- # A different event type that happens to carry a usage.cost_usd must be ignored.
+ assert boundary["status"] == "verified"
+ assert result == type(result)(Decimal("0.10"), ("missing_ci_capture:child",))
+
+
+def test_historical_boundary_excludes_later_parent_turn_cost(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ parent_messages = _messages("first", "second")
+ parent = _save(store, "parent", parent_messages, {})
_write_events(
- events,
+ parent / "context-intelligence" / "events.jsonl",
[
- {"event": "other:event", "data": {"usage": {"cost_usd": "9.99"}}},
- _llm_response("0.04"),
+ _submit("parent", "first"), _response("parent", "0.10"),
+ _submit("parent", "second"), _response("parent", "0.20"),
],
)
- assert sum_prior_cost_usd(events) == Decimal("0.04")
+ _, boundary = _fork(store, "parent", "child", parent_messages, {}, 1)
-# --------------------------------------------------------------------------
-# restore_session_cost
-# --------------------------------------------------------------------------
+ assert boundary["status"] == "verified"
+ assert boundary["cumulative_cost_usd_by_turn"] == ["0.10"]
-def test_restore_registers_history_contributor(tmp_path):
- events = tmp_path / "events.jsonl"
- _write_events(events, [_llm_response("0.10"), _llm_response("0.05")])
+def test_nested_boundary_survives_parent_rewrite_and_removed_ancestors(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ a_messages = _messages("a-one")
+ a = _save(store, "a", a_messages, {})
+ _write_events(
+ a / "context-intelligence" / "events.jsonl",
+ [_submit("a", "a-one"), _response("a", "0.10")],
+ )
+ b, b_boundary = _fork(store, "a", "b", a_messages, {}, 1)
+ assert b_boundary["status"] == "verified"
+
+ b_messages = _messages("a-one", "b-two")
+ store.save(
+ "b",
+ b_messages,
+ {
+ "session_id": "b",
+ "parent_id": "a",
+ "forked_from_turn": 1,
+ "fork_cost_boundary": b_boundary,
+ },
+ )
+ _write_events(
+ b / "context-intelligence" / "events.jsonl",
+ [_submit("b", "b-two"), _response("b", "0.30")],
+ )
+ c, c_boundary = _fork(
+ store,
+ "b",
+ "c",
+ b_messages,
+ store.get_metadata("b"),
+ 2,
+ )
+ assert c_boundary["cumulative_cost_usd_by_turn"] == ["0.10", "0.40"]
+ _write_events(c / "context-intelligence" / "events.jsonl", [_response("c", "0.40")])
+
+ before = restore_fork_lineage_cost(MagicMock(session_state={}), session_id="c", session_dir=c)
+ store.save("b", _messages("rewritten"), store.get_metadata("b"))
+ rewritten_boundary = build_fork_cost_boundary(
+ parent_dir=b,
+ parent_id="b",
+ parent_messages=_messages("rewritten"),
+ parent_metadata=store.get_metadata("b"),
+ fork_turn=1,
+ child_messages=_messages("rewritten"),
+ )
+ assert rewritten_boundary["status"] == "unavailable"
+ assert "fork_prefix_changed" in rewritten_boundary["reasons"]
+ for path in (a, b):
+ shutil.rmtree(path)
+ after = restore_fork_lineage_cost(MagicMock(session_state={}), session_id="c", session_dir=c)
- coordinator = MagicMock()
- registered = {}
+ assert before == after == type(before)(Decimal("0.80"), ())
- def capture_register(channel, name, callback):
- registered[(channel, name)] = callback
- coordinator.register_contributor = capture_register
+def test_resume_uses_snapshot_after_child_rewrite_but_new_fork_is_unavailable(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ parent_messages = _messages("parent")
+ parent = _save(store, "parent", parent_messages, {})
+ _write_events(
+ parent / "context-intelligence" / "events.jsonl",
+ [_submit("parent", "parent"), _response("parent", "0.10")],
+ )
+ child, boundary = _fork(store, "parent", "child", parent_messages, {}, 1)
+ child_messages = _messages("parent", "child")
+ child_metadata = store.get_metadata("child")
+ store.save("child", child_messages, child_metadata)
+ _write_events(
+ child / "context-intelligence" / "events.jsonl",
+ [_submit("child", "child"), _response("child", "0.20")],
+ )
- total = restore_session_cost(coordinator, "sess-abc", events)
+ before_rewrite = restore_fork_lineage_cost(
+ MagicMock(session_state={}), session_id="child", session_dir=child
+ )
+ rewritten_messages = _messages("rewritten")
+ store.save("child", rewritten_messages, child_metadata)
+ after_rewrite = restore_fork_lineage_cost(
+ MagicMock(session_state={}), session_id="child", session_dir=child
+ )
+ _, new_boundary = _fork(
+ store,
+ "child",
+ "new-child",
+ rewritten_messages,
+ store.get_metadata("child"),
+ 1,
+ )
- assert total == Decimal("0.15")
- key = ("session.cost", "history:sess-abc")
- assert key in registered
- # Contributor payload matches the provider modules' stringified shape.
- assert registered[key]() == {"cost_usd": "0.15"}
+ assert boundary["status"] == "verified"
+ assert before_rewrite == after_rewrite == type(before_rewrite)(Decimal("0.30"), ())
+ assert new_boundary["status"] == "unavailable"
+ assert "fork_prefix_changed" in new_boundary["reasons"]
-def test_restore_no_events_registers_nothing(tmp_path):
- coordinator = MagicMock()
- total = restore_session_cost(coordinator, "sess-x", tmp_path / "missing.jsonl")
- assert total is None
- coordinator.register_contributor.assert_not_called()
+def test_old_fork_without_boundary_is_incomplete_not_guessed(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ child = _save(
+ store,
+ "old-child",
+ _messages("one"),
+ {"parent_id": "gone-parent", "forked_from_turn": 1},
+ )
+ _write_events(child / "context-intelligence" / "events.jsonl", [_response("old-child", "0.20")])
+ result = restore_fork_lineage_cost(
+ MagicMock(session_state={}), session_id="old-child", session_dir=child
+ )
-def test_restore_zero_cost_registers_nothing(tmp_path):
- events = tmp_path / "events.jsonl"
- _write_events(events, [_llm_response("0")])
- coordinator = MagicMock()
- total = restore_session_cost(coordinator, "sess-x", events)
- assert total is None
- coordinator.register_contributor.assert_not_called()
+ assert result.total == Decimal("0.20")
+ assert result.diagnostics == ("missing_fork_cost_boundary",)
-def test_restore_swallows_registration_error(tmp_path):
- events = tmp_path / "events.jsonl"
- _write_events(events, [_llm_response("0.10")])
- coordinator = MagicMock()
- coordinator.register_contributor.side_effect = RuntimeError("boom")
- # Must never raise even if the kernel call fails.
- assert restore_session_cost(coordinator, "sess-x", events) is None
+def test_duplicate_prompt_and_corrupt_or_unscoped_ci_make_boundary_unavailable(
+ tmp_path: Path,
+) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ messages = _messages("same", "same")
+ parent = _save(store, "parent", messages, {})
+ events = parent / "context-intelligence" / "events.jsonl"
+ _write_events(events, [_submit("parent", "same"), _response("parent", "0.10")])
+ _, ambiguous = _fork(store, "parent", "ambiguous", messages, {}, 2)
+ assert ambiguous["status"] == "unavailable"
+ assert any(reason.startswith("missing_prompt_fence:parent") for reason in ambiguous["reasons"])
-def test_restore_and_fresh_provider_sum_without_double_count(tmp_path):
- """The history contributor + a fresh provider contributor sum correctly.
+ events.write_text(
+ json.dumps(_submit("parent", "same")) + "\n"
+ + json.dumps(_response("parent", "0.10")) + "\n"
+ + "{unfinished",
+ encoding="utf-8",
+ )
+ _, corrupt = _fork(store, "parent", "corrupt", _messages("same"), {}, 1)
+ assert corrupt["status"] == "unavailable"
+ assert "ci_incomplete_event:parent" in corrupt["reasons"]
- Mirrors resume: the provider re-mounts with a zeroed accumulator and only
- counts NEW turns; the history contributor supplies the pre-resume total.
- Together they reproduce the true cumulative cost with no double counting.
- """
- events = tmp_path / "events.jsonl"
- _write_events(events, [_llm_response("0.10"), _llm_response("0.05")])
- contributors = []
- coordinator = MagicMock()
- coordinator.register_contributor = lambda ch, name, cb: contributors.append(cb)
+def test_resumed_prompt_cycle_is_not_blindly_associated_by_ordinal(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ messages = _messages("repeat", "repeat")
+ parent = _save(store, "parent", messages, {})
+ _write_events(
+ parent / "context-intelligence" / "events.jsonl",
+ [
+ _submit("parent", "repeat"), _response("parent", "0.10"),
+ _submit("parent", "repeat"), _response("parent", "0.20"),
+ _submit("parent", "repeat"), _response("parent", "0.30"),
+ ],
+ )
- restore_session_cost(coordinator, "sess-abc", events) # history: 0.15
+ _, boundary = _fork(store, "parent", "child", messages, {}, 2)
+
+ assert boundary["status"] == "unavailable"
+ assert any(reason.startswith("missing_prompt_fence:parent") for reason in boundary["reasons"])
+
+
+def test_restore_excludes_unscoped_and_foreign_costs_and_is_idempotent(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ child = _save(
+ store,
+ "child",
+ _messages("one"),
+ {
+ "parent_id": "parent",
+ "forked_from_turn": 1,
+ "fork_cost_boundary": {
+ "version": 1,
+ "status": "verified",
+ "inherited_turn_count": 1,
+ "cumulative_cost_usd_by_turn": ["0.10"],
+ "prefix_fingerprint": "",
+ "provenance": {"owner": "parent", "fence": {}},
+ },
+ },
+ )
+ metadata = store.get_metadata("child")
+ metadata["fork_cost_boundary"]["prefix_fingerprint"] = _prefix_fingerprint(
+ _messages("one"), 1
+ )
+ store.save("child", _messages("one"), metadata)
+ _write_events(
+ child / "context-intelligence" / "events.jsonl",
+ [
+ _response("child", "0.20"),
+ {"event": "llm:response", "data": {"usage": {"cost_usd": "6.00"}}},
+ _response("foreign", "9.00"),
+ ],
+ )
+ coordinator = MagicMock(session_state={})
+ callbacks = _registered(coordinator)
- # Simulate a fresh per-mount provider contributor that recorded one new turn.
- contributors.append(lambda: {"cost_usd": "0.07"})
+ first = restore_fork_lineage_cost(coordinator, session_id="child", session_dir=child)
+ second = restore_fork_lineage_cost(coordinator, session_id="child", session_dir=child)
- # Reproduce how collect_contributions + sum_cost_usd aggregate the channel:
- # every registered contributor is summed (str or Decimal payloads accepted).
- total = sum(Decimal(str(cb()["cost_usd"])) for cb in contributors)
- assert total == Decimal("0.22")
+ assert first.total == second.total == Decimal("0.30")
+ assert "ci_unscoped_event:child" in first.diagnostics
+ assert "foreign_cost:child" in first.diagnostics
+ assert len(callbacks) == 1
-def test_ci_cost_scopes_parent_and_skips_non_finite_values(tmp_path):
- events = tmp_path / "events.jsonl"
- _write_events(events, [
- {"event": "llm:response", "data": {"session_id": "root", "usage": {"cost_usd": "0.12"}}},
- {"event": "llm:response", "data": {"session_id": "child", "usage": {"cost_usd": "4.00"}}},
- _llm_response("NaN"), _llm_response("Infinity"), _llm_response("-2"),
- ])
- coordinator = MagicMock()
- assert restore_session_cost(coordinator, "root", events) == Decimal("0.12")
-
-
-def test_ci_relocation_takes_precedence_over_local_and_legacy_cost_logs(tmp_path, monkeypatch):
- from amplifier_app_cli.cost_history import session_events_path
-
- session_dir = tmp_path / "projects" / "workspace-slug" / "sessions" / "root"
- session_dir.mkdir(parents=True)
- _write_events(session_dir / "events.jsonl", [_llm_response("9")])
- local = session_dir / "context-intelligence"
- local.mkdir()
- _write_events(local / "events.jsonl", [_llm_response("5")])
- relocated_root = tmp_path / "ci-captures"
- relocated = relocated_root / "workspace-slug" / "sessions" / "root" / "context-intelligence"
- relocated.mkdir(parents=True)
- _write_events(relocated / "events.jsonl", [_llm_response("0.25")])
- monkeypatch.setenv("AMPLIFIER_CONTEXT_INTELLIGENCE_BASE_PATH", str(relocated_root))
- selected = session_events_path(session_dir)
- assert selected == relocated / "events.jsonl"
- assert restore_session_cost(MagicMock(), "root", selected) == Decimal("0.25")
- monkeypatch.delenv("AMPLIFIER_CONTEXT_INTELLIGENCE_BASE_PATH")
- assert session_events_path(session_dir) == local / "events.jsonl"
- (local / "events.jsonl").unlink()
- assert session_events_path(session_dir) == session_dir / "events.jsonl"
-
-
-def test_ci_cost_helper_infers_session_id_above_capture_subdirectory(tmp_path):
- capture = tmp_path / "root" / "context-intelligence"
- capture.mkdir(parents=True)
- events = capture / "events.jsonl"
- _write_events(events, [
- {"event": "llm:response", "data": {"session_id": "root", "usage": {"cost_usd": "0.17"}}},
- {"event": "llm:response", "data": {"session_id": "other", "usage": {"cost_usd": "9"}}},
- ])
- assert sum_prior_cost_usd(events) == Decimal("0.17")
+def test_invalid_oversized_boundary_and_unstable_capture_are_explicit(tmp_path: Path, monkeypatch) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ messages = _messages("one")
+ parent = _save(store, "parent", messages, {})
+ _write_events(
+ parent / "context-intelligence" / "events.jsonl",
+ [_submit("parent", "one"), _response("parent", "0.10")],
+ )
+ stamps = iter(((1, 1, 1), (1, 1, 2)))
+ monkeypatch.setattr(
+ "amplifier_app_cli.cost_history._capture_stamp", lambda _path: next(stamps)
+ )
+ _, boundary = _fork(store, "parent", "unstable", messages, {}, 1)
+ assert boundary["status"] == "unavailable"
+ assert "unstable_ci_capture:parent" in boundary["reasons"]
+
+ child = _save(
+ store,
+ "oversized",
+ messages,
+ {
+ "parent_id": "parent",
+ "forked_from_turn": 1,
+ "fork_cost_boundary": {"version": 1, "status": "verified", "padding": "x" * 70_000},
+ },
+ )
+ result = restore_fork_lineage_cost(
+ MagicMock(session_state={}), session_id="oversized", session_dir=child
+ )
+ assert "oversized_fork_cost_boundary" in result.diagnostics
+
+ invalid = _save(
+ store,
+ "invalid",
+ messages,
+ {
+ "parent_id": "parent",
+ "forked_from_turn": 1,
+ "fork_cost_boundary": {"version": 1, "status": "verified"},
+ },
+ )
+ invalid_result = restore_fork_lineage_cost(
+ MagicMock(session_state={}), session_id="invalid", session_dir=invalid
+ )
+ assert "invalid_fork_cost_boundary" in invalid_result.diagnostics
+
+
+def test_boundary_uses_selected_relocated_capture_only(tmp_path: Path, monkeypatch) -> None:
+ store = SessionStore(tmp_path / "projects" / "slug" / "sessions")
+ messages = _messages("one")
+ parent = _save(store, "parent", messages, {})
+ _write_events(
+ parent / "context-intelligence" / "events.jsonl",
+ [_submit("parent", "one"), _response("parent", "9.00")],
+ )
+ relocated = tmp_path / "relocated" / "slug" / "sessions" / "parent" / "context-intelligence"
+ _write_events(
+ relocated / "events.jsonl",
+ [_submit("parent", "one"), _response("parent", "0.10")],
+ )
+ monkeypatch.setenv("AMPLIFIER_CONTEXT_INTELLIGENCE_BASE_PATH", str(tmp_path / "relocated"))
+
+ _, boundary = _fork(store, "parent", "child", messages, {}, 1)
+
+ assert boundary["status"] == "verified"
+ assert boundary["cumulative_cost_usd_by_turn"] == ["0.10"]
+
+
+def test_missing_selected_child_capture_keeps_verified_snapshot_and_is_incomplete(
+ tmp_path: Path, monkeypatch
+) -> None:
+ store = SessionStore(tmp_path / "projects" / "slug" / "sessions")
+ parent_messages = _messages("parent")
+ parent = _save(store, "parent", parent_messages, {})
+ relocated = tmp_path / "relocated" / "slug" / "sessions"
+ _write_events(
+ relocated / "parent" / "context-intelligence" / "events.jsonl",
+ [_submit("parent", "parent"), _response("parent", "0.10")],
+ )
+ monkeypatch.setenv("AMPLIFIER_CONTEXT_INTELLIGENCE_BASE_PATH", str(tmp_path / "relocated"))
+ child, boundary = _fork(store, "parent", "child", parent_messages, {}, 1)
+ _write_events(
+ child / "context-intelligence" / "events.jsonl",
+ [_response("child", "9.99")],
+ )
+
+ result = restore_fork_lineage_cost(MagicMock(session_state={}), session_id="child", session_dir=child)
+
+ assert boundary["status"] == "verified"
+ assert result.total == Decimal("0.10")
+ assert result.diagnostics == ("missing_ci_capture:child",)
+
+
+def test_native_reminder_turn_carries_cost_and_hashes_foundation_prefix(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ messages = [
+ {"role": "user", "content": "human one"},
+ {"role": "assistant", "content": "answer one"},
+ {
+ "role": "user",
+ "content": "remember",
+ "metadata": {"ephemeral": True, "persisted": True},
+ },
+ {"role": "assistant", "content": "hook output"},
+ {"role": "user", "content": "human two"},
+ {"role": "assistant", "content": "answer two"},
+ ]
+ parent = _save(store, "parent", messages, {})
+ _write_events(
+ parent / "context-intelligence" / "events.jsonl",
+ [_submit("parent", "human one"), _response("parent", "0.10")],
+ )
+ child_messages = slice_to_turn(messages, 2, handle_orphaned_tools="complete")
+
+ boundary = build_fork_cost_boundary(
+ parent_dir=parent,
+ parent_id="parent",
+ parent_messages=messages,
+ parent_metadata={},
+ fork_turn=2,
+ child_messages=child_messages,
+ )
+
+ canonical = [sanitize_message(message) for message in child_messages]
+ expected_prefix = slice_to_turn(canonical, 2, handle_orphaned_tools="complete")
+ expected_fingerprint = hashlib.sha256(
+ json.dumps(
+ expected_prefix,
+ ensure_ascii=False,
+ sort_keys=True,
+ separators=(",", ":"),
+ allow_nan=False,
+ ).encode("utf-8")
+ ).hexdigest()
+ assert boundary["status"] == "verified"
+ assert boundary["cumulative_cost_usd_by_turn"] == ["0.10", "0.10"]
+ assert boundary["prefix_fingerprint"] == expected_fingerprint
+ assert boundary["warnings"] == ["unmapped_non_anchor_output:parent:2"]
+
+
+def test_fork_rejects_intra_prompt_cutoff_but_accepts_full_reminder_prefix(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ messages = [
+ {"role": "user", "content": "human one"},
+ {"role": "assistant", "content": "answer one"},
+ {
+ "role": "user",
+ "content": "remember",
+ "metadata": {"ephemeral": True, "persisted": True},
+ },
+ {"role": "assistant", "content": "hook output"},
+ {"role": "user", "content": "human two"},
+ {"role": "assistant", "content": "answer two"},
+ ]
+ parent = _save(store, "parent", messages, {})
+ _write_events(
+ parent / "context-intelligence" / "events.jsonl",
+ [
+ _submit("parent", "human one"),
+ _response("parent", "0.10"),
+ _response("parent", "0.20"),
+ _submit("parent", "human two"),
+ _response("parent", "0.30"),
+ ],
+ )
+
+ cutoff = build_fork_cost_boundary(
+ parent_dir=parent,
+ parent_id="parent",
+ parent_messages=messages,
+ parent_metadata={},
+ fork_turn=1,
+ child_messages=slice_to_turn(messages, 1, handle_orphaned_tools="complete"),
+ )
+ through_reminder = build_fork_cost_boundary(
+ parent_dir=parent,
+ parent_id="parent",
+ parent_messages=messages,
+ parent_metadata={},
+ fork_turn=2,
+ child_messages=slice_to_turn(messages, 2, handle_orphaned_tools="complete"),
+ )
+
+ assert cutoff["status"] == "unavailable"
+ assert cutoff["reasons"] == ["unprovable_intra_prompt_cutoff"]
+ assert through_reminder["status"] == "verified"
+ assert through_reminder["cumulative_cost_usd_by_turn"] == ["0.30", "0.30"]
+
+
+def test_trailing_reminder_without_assistant_output_does_not_block_fork(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ messages = [
+ {"role": "user", "content": "human"},
+ {"role": "assistant", "content": "answer"},
+ {
+ "role": "user",
+ "content": "remember",
+ "metadata": {"ephemeral": True, "persisted": True},
+ },
+ ]
+ parent = _save(store, "parent", messages, {})
+ _write_events(
+ parent / "context-intelligence" / "events.jsonl",
+ [_submit("parent", "human"), _response("parent", "0.10")],
+ )
+
+ boundary = build_fork_cost_boundary(
+ parent_dir=parent,
+ parent_id="parent",
+ parent_messages=messages,
+ parent_metadata={},
+ fork_turn=1,
+ child_messages=slice_to_turn(messages, 1, handle_orphaned_tools="complete"),
+ )
+
+ assert boundary["status"] == "verified"
+ assert boundary["cumulative_cost_usd_by_turn"] == ["0.10"]
+
+
+def test_nested_fork_rejects_earlier_inherited_intra_prompt_cutoff(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ a_messages = [
+ {"role": "user", "content": "a one"},
+ {"role": "assistant", "content": "a answer"},
+ {
+ "role": "user",
+ "content": "remember",
+ "metadata": {"ephemeral": True, "persisted": True},
+ },
+ {"role": "assistant", "content": "hook output"},
+ {"role": "user", "content": "a two"},
+ {"role": "assistant", "content": "a second answer"},
+ ]
+ a = _save(store, "a", a_messages, {})
+ _write_events(
+ a / "context-intelligence" / "events.jsonl",
+ [
+ _submit("a", "a one"),
+ _response("a", "0.10"),
+ _response("a", "0.20"),
+ _submit("a", "a two"),
+ _response("a", "0.30"),
+ ],
+ )
+ b_messages = slice_to_turn(a_messages, 2, handle_orphaned_tools="complete")
+ b_boundary = build_fork_cost_boundary(
+ parent_dir=a,
+ parent_id="a",
+ parent_messages=a_messages,
+ parent_metadata={},
+ fork_turn=2,
+ child_messages=b_messages,
+ )
+ b_messages.extend(({"role": "user", "content": "b"}, {"role": "assistant", "content": "b answer"}))
+ b = _save(
+ store,
+ "b",
+ b_messages,
+ {"parent_id": "a", "forked_from_turn": 2, "fork_cost_boundary": b_boundary},
+ )
+
+ boundary = build_fork_cost_boundary(
+ parent_dir=b,
+ parent_id="b",
+ parent_messages=b_messages,
+ parent_metadata=store.get_metadata("b"),
+ fork_turn=1,
+ child_messages=slice_to_turn(b_messages, 1, handle_orphaned_tools="complete"),
+ )
+
+ assert b_boundary["status"] == "verified"
+ assert boundary["status"] == "unavailable"
+ assert boundary["reasons"] == ["unprovable_intra_prompt_cutoff"]
+
+
+def test_nested_fork_preserves_an_inherited_reminder_snapshot_and_fingerprint(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ a_messages = [
+ {"role": "user", "content": "a"},
+ {"role": "assistant", "content": "a answer"},
+ {
+ "role": "user",
+ "content": "remember",
+ "metadata": {"ephemeral": True, "persisted": True},
+ },
+ {"role": "assistant", "content": "hook output"},
+ ]
+ a = _save(store, "a", a_messages, {})
+ _write_events(
+ a / "context-intelligence" / "events.jsonl",
+ [_submit("a", "a"), _response("a", "0.10")],
+ )
+ b_messages = slice_to_turn(a_messages, 2, handle_orphaned_tools="complete")
+ b_boundary = build_fork_cost_boundary(
+ parent_dir=a,
+ parent_id="a",
+ parent_messages=a_messages,
+ parent_metadata={},
+ fork_turn=2,
+ child_messages=b_messages,
+ )
+ b = _save(
+ store,
+ "b",
+ b_messages,
+ {
+ "parent_id": "a",
+ "forked_from_turn": 2,
+ "fork_cost_boundary": b_boundary,
+ },
+ )
+ b_messages = [*b_messages, {"role": "user", "content": "b"}, {"role": "assistant", "content": "b answer"}]
+ store.save("b", b_messages, store.get_metadata("b"))
+ _write_events(
+ b / "context-intelligence" / "events.jsonl",
+ [_submit("b", "b"), _response("b", "0.20")],
+ )
+ c_messages = slice_to_turn(b_messages, 3, handle_orphaned_tools="complete")
+
+ c_boundary = build_fork_cost_boundary(
+ parent_dir=b,
+ parent_id="b",
+ parent_messages=b_messages,
+ parent_metadata=store.get_metadata("b"),
+ fork_turn=3,
+ child_messages=c_messages,
+ )
+
+ assert b_boundary["status"] == "verified"
+ assert b_boundary["cumulative_cost_usd_by_turn"] == ["0.10", "0.10"]
+ assert c_boundary["status"] == "verified"
+ assert c_boundary["cumulative_cost_usd_by_turn"] == ["0.10", "0.10", "0.30"]
+ assert c_boundary["prefix_fingerprint"] == _prefix_fingerprint(c_messages, 3)
+
+
+def test_multimodal_prompt_uses_only_first_text_block_and_unsupported_anchor_is_unavailable(
+ tmp_path: Path,
+) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ supported = [
+ {
+ "role": "user",
+ "content": [
+ {"type": "text", "text": "caption"},
+ {"type": "image", "source": {"type": "base64", "data": "..."}},
+ {"type": "text", "text": "trailing text is not submitted"},
+ ],
+ },
+ {"role": "assistant", "content": "answer"},
+ ]
+ parent = _save(store, "supported", supported, {})
+ _write_events(
+ parent / "context-intelligence" / "events.jsonl",
+ [_submit("supported", "caption"), _response("supported", "0.10")],
+ )
+ verified = build_fork_cost_boundary(
+ parent_dir=parent,
+ parent_id="supported",
+ parent_messages=supported,
+ parent_metadata={},
+ fork_turn=1,
+ child_messages=slice_to_turn(supported, 1, handle_orphaned_tools="complete"),
+ )
+ assert verified["status"] == "verified"
+ assert verified["cumulative_cost_usd_by_turn"] == ["0.10"]
+
+ unsupported = [
+ {"role": "user", "content": [{"type": "image", "source": {"type": "base64"}}]},
+ {"role": "assistant", "content": "answer"},
+ ]
+ unsupported_parent = _save(store, "unsupported", unsupported, {})
+ unavailable = build_fork_cost_boundary(
+ parent_dir=unsupported_parent,
+ parent_id="unsupported",
+ parent_messages=unsupported,
+ parent_metadata={},
+ fork_turn=1,
+ child_messages=slice_to_turn(unsupported, 1, handle_orphaned_tools="complete"),
+ )
+ assert unavailable["status"] == "unavailable"
+ assert "unsupported_prompt_anchor:unsupported:1" in unavailable["reasons"]
+
+
+def test_nested_owner_suffix_ignores_inherited_yes_and_accepts_exact_repeats(tmp_path: Path) -> None:
+ store = SessionStore(tmp_path / "sessions")
+ a_messages = _messages("yes")
+ a = _save(store, "a", a_messages, {})
+ _write_events(
+ a / "context-intelligence" / "events.jsonl",
+ [_submit("a", "yes"), _response("a", "0.10")],
+ )
+ b, b_boundary = _fork(store, "a", "b", a_messages, {}, 1)
+ assert b_boundary["status"] == "verified"
+
+ one_owned = _messages("yes", "yes")
+ store.save(
+ "b",
+ one_owned,
+ {
+ "session_id": "b",
+ "parent_id": "a",
+ "forked_from_turn": 1,
+ "fork_cost_boundary": b_boundary,
+ },
+ )
+ _write_events(
+ b / "context-intelligence" / "events.jsonl",
+ [_submit("b", "yes"), _response("b", "0.20")],
+ )
+ _c, c_boundary = _fork(store, "b", "c", one_owned, store.get_metadata("b"), 2)
+ assert c_boundary["status"] == "verified"
+ assert c_boundary["cumulative_cost_usd_by_turn"] == ["0.10", "0.30"]
+
+ repeated_owned = _messages("yes", "yes", "yes")
+ store.save(
+ "b",
+ repeated_owned,
+ {
+ "session_id": "b",
+ "parent_id": "a",
+ "forked_from_turn": 1,
+ "fork_cost_boundary": b_boundary,
+ },
+ )
+ _write_events(
+ b / "context-intelligence" / "events.jsonl",
+ [
+ _submit("b", "yes"), _response("b", "0.20"),
+ _submit("b", "yes"), _response("b", "0.30"),
+ ],
+ )
+ _d, d_boundary = _fork(store, "b", "d", repeated_owned, store.get_metadata("b"), 3)
+ assert d_boundary["status"] == "verified"
+ assert d_boundary["cumulative_cost_usd_by_turn"] == ["0.10", "0.30", "0.60"]
diff --git a/tests/test_session_runner.py b/tests/test_session_runner.py
index b7b528c6..a3d4b5b4 100644
--- a/tests/test_session_runner.py
+++ b/tests/test_session_runner.py
@@ -1,7 +1,9 @@
"""Tests for session_runner module - unified session initialization."""
import logging
+import json
import sys as _sys
+from decimal import Decimal
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
@@ -14,6 +16,7 @@
create_initialized_session,
register_session_spawning,
)
+from amplifier_app_cli.cost_history import ForkLineageCost
# ---------------------------------------------------------------------------
# Helpers shared across tests
@@ -49,6 +52,86 @@ def _make_mock_session(initial_session_config=None):
return mock_sess
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+ ("metadata", "expects_lineage"),
+ [
+ ({"session_id": "fork", "parent_id": "root", "forked_from_turn": 1}, True),
+ ({"session_id": "agent-child", "parent_id": "root"}, False),
+ ],
+)
+async def test_resume_uses_lineage_cost_only_for_native_transcript_forks(
+ tmp_path, metadata, expects_lineage
+):
+ """A parent alone identifies an agent child, not a conversation fork."""
+ from contextlib import ExitStack
+
+ session_id = metadata["session_id"]
+ session_dir = tmp_path / session_id
+ session_dir.mkdir()
+ (session_dir / "metadata.json").write_text(json.dumps(metadata), encoding="utf-8")
+ mock_session = _make_mock_session()
+ cfg = _make_session_config(
+ session_id=session_id, initial_transcript=[{"role": "user", "content": "resume"}]
+ )
+ store = MagicMock(base_dir=tmp_path)
+ lineage_result = ForkLineageCost(Decimal("0.10"), ())
+
+ with ExitStack() as stack:
+ for patcher in _configurator_patches(mock_session):
+ stack.enter_context(patcher)
+ stack.enter_context(patch(f"{_MODULE}.SessionStore", return_value=store))
+ restore_lineage = stack.enter_context(
+ patch(
+ "amplifier_app_cli.cost_history.restore_fork_lineage_cost",
+ return_value=lineage_result,
+ )
+ )
+ restore_ordinary = stack.enter_context(
+ patch("amplifier_app_cli.cost_history.restore_session_cost")
+ )
+ await create_initialized_session(cfg, MagicMock())
+
+ assert restore_lineage.called is expects_lineage
+ assert restore_ordinary.called is not expects_lineage
+
+
+@pytest.mark.anyio
+async def test_resume_warns_when_a_fork_cost_segment_is_incomplete(tmp_path) -> None:
+ from contextlib import ExitStack
+
+ session_id = "fork"
+ session_dir = tmp_path / session_id
+ session_dir.mkdir()
+ (session_dir / "metadata.json").write_text(
+ json.dumps({"session_id": session_id, "parent_id": "root", "forked_from_turn": 1}),
+ encoding="utf-8",
+ )
+ mock_session = _make_mock_session()
+ cfg = _make_session_config(
+ session_id=session_id, initial_transcript=[{"role": "user", "content": "resume"}]
+ )
+ store = MagicMock(base_dir=tmp_path)
+ console = MagicMock()
+
+ with ExitStack() as stack:
+ for patcher in _configurator_patches(mock_session):
+ stack.enter_context(patcher)
+ stack.enter_context(patch(f"{_MODULE}.SessionStore", return_value=store))
+ stack.enter_context(
+ patch(
+ "amplifier_app_cli.cost_history.restore_fork_lineage_cost",
+ return_value=ForkLineageCost(Decimal("0.10"), ("missing_ci_capture:fork",)),
+ )
+ )
+ await create_initialized_session(cfg, console)
+
+ console.print.assert_called_once_with(
+ "[yellow]Warning:[/yellow] cumulative fork cost history is incomplete; "
+ "only verified available CI cost segments were restored."
+ )
+
+
# ---------------------------------------------------------------------------
# SessionConfig tests
# ---------------------------------------------------------------------------
diff --git a/tests/test_shared_root_lifecycle.py b/tests/test_shared_root_lifecycle.py
index 264670de..d26c68b1 100644
--- a/tests/test_shared_root_lifecycle.py
+++ b/tests/test_shared_root_lifecycle.py
@@ -20,6 +20,7 @@
SharedRootSession,
read_shared_root,
)
+from amplifier_foundation.session import slice_to_turn
pytestmark = pytest.mark.skipif(
sys.platform == "win32",
@@ -77,6 +78,13 @@ def test_session_fork_reads_latest_native_history_over_legacy_checkpoint(
_messages("stale projection"),
{"session_id": "shared-root", "bundle": "bundle:stale"},
)
+ parent_capture = native.base_dir / "shared-root" / "context-intelligence" / "events.jsonl"
+ parent_capture.parent.mkdir()
+ parent_capture.write_text(
+ '{"event":"prompt:submit","data":{"session_id":"shared-root","prompt":"stale projection"}}\n'
+ '{"event":"llm:response","data":{"session_id":"shared-root","usage":{"cost_usd":"0.10"}}}\n',
+ encoding="utf-8",
+ )
result = CliRunner().invoke(
_session_cli(native, monkeypatch),
@@ -89,6 +97,46 @@ def test_session_fork_reads_latest_native_history_over_legacy_checkpoint(
assert metadata["bundle"] == "bundle:stale"
assert metadata["parent_id"] == "shared-root"
assert metadata["forked_from_turn"] == 1
+ assert metadata["fork_cost_boundary"]["status"] == "verified"
+ assert metadata["fork_cost_boundary"]["cumulative_cost_usd_by_turn"] == ["0.10"]
+
+
+def test_session_fork_uses_native_reminder_turns_for_verified_cost_metadata(
+ tmp_path: Path, monkeypatch
+) -> None:
+ native = _isolate_state(tmp_path, monkeypatch)
+ messages = [
+ {"role": "user", "content": "human one"},
+ {"role": "assistant", "content": "answer one"},
+ {
+ "role": "user",
+ "content": "remember",
+ "metadata": {"ephemeral": True, "persisted": True},
+ },
+ {"role": "assistant", "content": "hook output"},
+ {"role": "user", "content": "human two"},
+ {"role": "assistant", "content": "answer two"},
+ ]
+ native.save("mixed-root", messages, {"session_id": "mixed-root"})
+ capture = native.base_dir / "mixed-root" / "context-intelligence" / "events.jsonl"
+ capture.parent.mkdir()
+ capture.write_text(
+ '{"event":"prompt:submit","data":{"session_id":"mixed-root","prompt":"human one"}}\n'
+ '{"event":"llm:response","data":{"session_id":"mixed-root","usage":{"cost_usd":"0.10"}}}\n',
+ encoding="utf-8",
+ )
+
+ result = CliRunner().invoke(
+ _session_cli(native, monkeypatch),
+ ["session", "fork", "mixed-root", "--at-turn", "2", "--name", "mixed-child"],
+ )
+
+ assert result.exit_code == 0, result.output
+ transcript, metadata = native.load("mixed-child")
+ assert transcript == slice_to_turn(messages, 2, handle_orphaned_tools="complete")
+ assert metadata["forked_from_turn"] == 2
+ assert metadata["fork_cost_boundary"]["status"] == "verified"
+ assert metadata["fork_cost_boundary"]["cumulative_cost_usd_by_turn"] == ["0.10", "0.10"]
def test_real_foundation_lock_writes_native_metadata_without_checkpoint(
@@ -271,6 +319,7 @@ def get_capability(name: str):
assert transcript[0]["content"] == "live authority"
assert metadata["parent_id"] == "shared-root"
assert metadata["bundle"] == "bundle:held"
+ assert metadata["fork_cost_boundary"]["status"] == "unavailable"
root_handle.read.assert_called_once_with(native)
native.save(
@@ -282,6 +331,95 @@ def get_capability(name: str):
assert (await processor._fork_session("1 shared-root_worker")).startswith("Error:")
+@pytest.mark.asyncio
+async def test_interactive_fork_persists_verified_boundary_from_live_context(
+ tmp_path: Path, monkeypatch
+) -> None:
+ native = _isolate_state(tmp_path, monkeypatch)
+ messages = _messages("interactive cost")
+ native.save("interactive-root", messages, {"session_id": "interactive-root"})
+ capture = native.base_dir / "interactive-root" / "context-intelligence" / "events.jsonl"
+ capture.parent.mkdir()
+ capture.write_text(
+ '{"event":"prompt:submit","data":{"session_id":"interactive-root","prompt":"interactive cost"}}\n'
+ '{"event":"llm:response","data":{"session_id":"interactive-root","usage":{"cost_usd":"0.10"}}}\n',
+ encoding="utf-8",
+ )
+ context = type("Context", (), {"get_messages": AsyncMock(return_value=messages)})()
+ root_handle = MagicMock()
+ root_handle.read.return_value = (messages, {"session_id": "interactive-root"})
+ session = type("Session", (), {})()
+ session.coordinator = type(
+ "Coordinator",
+ (),
+ {
+ "session_id": "interactive-root",
+ "session_state": {},
+ "get": staticmethod(lambda name: context if name == "context" else None),
+ "get_capability": staticmethod(
+ lambda name: root_handle if name == "cli.shared_root_state" else None
+ ),
+ },
+ )()
+ main_module = import_module("amplifier_app_cli.main")
+ store_module = import_module("amplifier_app_cli.session_store")
+ monkeypatch.setattr(store_module, "SessionStore", lambda: native)
+
+ result = await main_module.CommandProcessor(session, "bundle:anchors")._fork_session(
+ "1 interactive-child"
+ )
+
+ assert not result.startswith("Error"), result
+ boundary = native.get_metadata("interactive-child")["fork_cost_boundary"]
+ assert boundary["status"] == "verified"
+ assert boundary["cumulative_cost_usd_by_turn"] == ["0.10"]
+
+
+@pytest.mark.asyncio
+async def test_interactive_legacy_fork_never_includes_parent_event_logs(
+ tmp_path: Path, monkeypatch
+) -> None:
+ """The non-shared `/fork` branch writes boundary metadata without copying logs."""
+ native = _isolate_state(tmp_path, monkeypatch)
+ messages = _messages("legacy authority")
+ native.save("legacy-root", messages, {"session_id": "legacy-root"})
+ parent = native.base_dir / "legacy-root"
+ (parent / "events.jsonl").write_text('{"event":"llm:response"}\n', encoding="utf-8")
+ capture = parent / "context-intelligence"
+ capture.mkdir()
+ capture_events = capture / "events.jsonl"
+ capture_events.write_text('{"event":"llm:response"}\n', encoding="utf-8")
+ before = (parent / "events.jsonl").read_bytes(), capture_events.read_bytes()
+
+ context = type("Context", (), {"get_messages": AsyncMock(return_value=messages)})()
+ session = type("Session", (), {})()
+ session.coordinator = type(
+ "Coordinator",
+ (),
+ {
+ "session_id": "legacy-root",
+ "session_state": {},
+ "get": staticmethod(lambda name: context if name == "context" else None),
+ "get_capability": staticmethod(lambda _name: None),
+ },
+ )()
+ main_module = import_module("amplifier_app_cli.main")
+ store_module = import_module("amplifier_app_cli.session_store")
+ monkeypatch.setattr(store_module, "SessionStore", lambda: native)
+ result = await main_module.CommandProcessor(session, "bundle:anchors")._fork_session(
+ "1 legacy-child"
+ )
+
+ assert not result.startswith("Error"), result
+ child = native.base_dir / "legacy-child"
+ assert not (child / "events.jsonl").exists()
+ assert not (child / "context-intelligence").exists()
+ assert (parent / "events.jsonl").read_bytes() == before[0]
+ assert capture_events.read_bytes() == before[1]
+ assert native.get_metadata("legacy-child")["fork_cost_boundary"]["status"] == "unavailable"
+ assert "Event history remains with its original owners." in result
+
+
def test_busy_shared_root_is_a_click_error_with_parseable_json_stdout(
tmp_path: Path, monkeypatch
) -> None:
@@ -359,9 +497,10 @@ def test_busy_shared_root_is_a_click_error_with_parseable_json_stdout(
assert "Shared root session is busy" in result.stderr
@pytest.mark.parametrize("skip_events", [False, True])
-def test_native_fork_preserves_legacy_log_option_without_copying_ci(tmp_path, monkeypatch, skip_events):
- import json
-
+@pytest.mark.parametrize("shared_lock", [False, True])
+def test_native_fork_leaves_all_event_logs_with_original_owners(
+ tmp_path, monkeypatch, skip_events, shared_lock
+):
native = _isolate_state(tmp_path, monkeypatch)
messages = [
{"role": "user", "content": "first", "timestamp": "2026-01-01T00:00:00Z"},
@@ -371,7 +510,9 @@ def test_native_fork_preserves_legacy_log_option_without_copying_ci(tmp_path, mo
native.save("root", messages, {"bundle": "bundle:anchors"})
parent = native.base_dir / "root"
legacy = parent / "events.jsonl"
- legacy.write_text(json.dumps({"event": "prompt:submit", "session_id": "root", "ts": "2026-01-01T00:00:00Z", "data": {}}) + "\n")
+ legacy.write_text(
+ '{"event":"prompt:submit","session_id":"root","ts":"2026-01-01T00:00:00Z","data":{}}\n'
+ )
capture = parent / "context-intelligence"
capture.mkdir()
(capture / "events.jsonl").write_text('{"event":"prompt:submit","data":{"session_id":"root"}}\n')
@@ -379,17 +520,20 @@ def test_native_fork_preserves_legacy_log_option_without_copying_ci(tmp_path, mo
args = ["session", "fork", "root", "--at-turn", "1", "--name", "forked"]
if skip_events:
args.append("--no-events")
+ monkeypatch.setattr(
+ session_commands, "_shared_root_platform_supported", lambda: shared_lock
+ )
result = CliRunner().invoke(_session_cli(native, monkeypatch), args)
assert result.exit_code == 0, result.output
child = native.base_dir / "forked"
assert native.load("forked")[0] == messages[:2]
+ assert native.load("forked")[1]["fork_cost_boundary"]["status"] == "unavailable"
assert not (child / "context-intelligence").exists()
assert (capture / "events.jsonl").read_bytes() == before
- assert (child / "events.jsonl").exists() is not skip_events
- if not skip_events:
- copied = json.loads((child / "events.jsonl").read_text())
- assert copied["session_id"] == "forked"
- assert copied["parent_session_id"] == "root"
+ assert not (child / "events.jsonl").exists()
+ assert "Event history remains with its original owners." in result.output
+ if skip_events:
+ assert "--no-events accepted for compatibility" in result.output
@pytest.mark.parametrize("primary", ["missing", "corrupt"])