Architecture position

Capability Extensions

Extend what an agent can do through capability packages, external evidence, persistent state, and standard protocols.

How do we add Skills, retrieval, memory, and external protocol capabilities to an agent?

Concept calibration

Memory

What it is: A mechanism for storing, retrieving, and using state across steps or sessions, with explicit write, read, and forgetting policies.
What it is not: It is not merely compressed chat history, and it should not put every old item back into context.

Context

What it is: The information visible to a model call, including instructions, conversation, tool schemas and results, retrieved evidence, and current state.
What it is not: It is not merely chat history, nor everything the model can remember permanently.

Observability

What it is: Use structured logs, traces, metrics, and version data to explain what happened during a system run.
What it is not: It is not just terminal output; logs without correlation IDs, state transitions, and privacy boundaries are hard to diagnose.

State, Sessions, Memory & Context Compression

Design session state, checkpoints, short- and long-term memory, compression, and replay.

One thing to completeMake long tasks pausable, resumable, compressible, and replayable.

Before you begin, Forge already hasThe accepted solution from Module 09, RAG, Retrieval & Knowledge Evaluation, is this chapter’s starting point.

After this chapter, Forge canPause, resume, and replay long tasks while controlling context growth.

Smallest recovery pointPersist explicit state and the latest checkpoint only; leave long-term memory off.

Download this chapter labPython 3.12 · pytest · Pydantic · SQLite · deterministic Mock

Stage capability chain

01Distinguish current state from historical facts and deterministically rebuild RunState from immutable events02Distinguish a session from an individual run and explicitly manage start, finish, and concurrency boundaries03Persist a minimal versioned checkpoint with integrity verification and reject corrupt or unsupported snapshots04Recover from a checkpoint plus tail events and prove that the result matches a full event replay05Separate recoverable runtime state from retrievable long-term memory with scope, expiry, and deletion semantics06Compress ordinary conversation while preserving instructions, permission decisions, and commitments verbatim with fidelity checks07Replay a run from an auditable event journal and propagate memory deletion to dependent derived-summary caches

Each step adds one verifiable capability; every later step begins from code and tests accepted in the previous step.

Concept calibration\n\nPut the concepts used in this module back inside their engineering boundaries before you work with the code.\n\n
\n\n### Memory memory\n\n- What it is: A mechanism for storing, retrieving, and using state across steps or sessions, with explicit write, read, conflict, retention, and forgetting policies.\n- What it is not: It is not merely a history summary, and it should not put every old record into each context; stored data becomes useful memory only when retrieved and applied correctly.\n- Relationship to adjacent concepts: State represents current progress, checkpoints preserve consistent snapshots, memory reuses information across boundaries, and context carries only the selected portion.\n- Where it lives in HeatStack Forge: Forge stores task state, user approvals, and longer-lived preferences separately, retrieves by version and scope, and supports pause, resume, and replay.\n- Typical misuse and correction: A common misuse writes every model output automatically. Set write thresholds, provenance, expiry, and conflict policy, then test that stale data cannot override current facts.\n\n### Context context\n\n- What it is: The information visible to one model call, including instructions, conversation excerpts, tool schemas and results, retrieved evidence, current state, and output constraints.\n- What it is not: It is not merely chat history or every item in a database; only information selected for the current call belongs to its context.\n- Relationship to adjacent concepts: RAG selects external evidence, memory preserves cross-step state, and tool schemas describe actions. All can supply context, but their responsibilities differ.\n- Where it lives in HeatStack Forge: Forge assembles context by task, permission, budget, and source priority while recording provenance and version for each item.\n- Typical misuse and correction: A common misuse equates more tokens with better understanding. Use selection, truncation, citation, and sensitive-data filters, then verify them on fixed fixtures.\n\n### Observability observability\n\n- What it is: Structured logs, traces, metrics, state transitions, and version information that explain a run and support failure diagnosis and version comparison.\n- What it is not: It is not merely terminal output or unlimited input recording; logs without correlation IDs, state transitions, or privacy boundaries are hard to diagnose and may leak data.\n- Relationship to adjacent concepts: The harness emits runtime events, the agent loop and tool calls form traces, evaluation aggregates metrics, and memory and recovery depend on versioned records.\n- Where it lives in HeatStack Forge: Forge creates an operation_id for each task and records stage, tool, latency, cost, error class, state change, and evidence path while filtering sensitive data.\n- Typical misuse and correction: A common misuse adds logs only after failure. Define events and correlation fields with the contract, then use fault injection to prove the timeline can be reconstructed.\n\n

STAGE 01

Distinguish current state from historical facts and deterministically rebuild RunState from immutable events

Start with a concrete problem

How to rebuild current state from an event sequence?

You are building a long-running task execution system that must record key facts during a run, such as run creation, step start, and permission revocation. These facts are immutable once they occur, but the current state (e.g., whether the run is still active, what the current step is, whether writing is allowed) changes as events are applied.

If you directly mutate a state object, you cannot audit how the state evolved or recover from a crash. Therefore, you need to treat events as an immutable history and rebuild state deterministically by replaying them.

In this stage, you will implement the core of event sourcing: define event and state models, and write an apply_event function that updates state for each event. You must also ensure the event sequence is contiguous and that no events are accepted after the run reaches a terminal state.

Make a prediction

Before implementing reduce_events, predict: if the event sequence has a gap (e.g., sequence 1 followed directly by sequence 3), how should the system handle it?

Ignore the gap and continue applying subsequent events
Raise an exception because the state cannot be reliably rebuilt
Automatically fill in the missing events
Log a warning but continue execution

Reasoning guide: The correct answer is to raise an exception. The event sequence must be contiguous because each event depends on the state produced by the previous event. A gap indicates incomplete history, so the system must refuse to rebuild rather than guess or ignore.

What it is

State is a projection of current progress, events are historical facts

State (RunState) represents the latest information about a run at a given moment, such as run status, current step, and write permission. It is derived from the event sequence, not stored independently.

Events (RunEvent) are immutable historical records, each containing a sequence number, type, and payload. Once created, events cannot be modified, so they can be safely replayed.

The apply_event function applies events one by one to the state, producing the current state. This process is deterministic: given the same event sequence, the same state is always produced.

What it is not

State is not history, and events are not mutable objects

State is not a list of all events; it contains only the information needed for the current moment. Historical events may be kept for audit, but state is a condensed projection.

Events are not mutable; once created, they cannot be changed. If an error is discovered, you cannot modify an old event; instead, you append a new event to correct it.

State cannot be read directly from history; it must be rebuilt by replaying events to ensure consistency and auditability.

How it relates to neighboring concepts

Events drive state transitions

Each event corresponds to a state transition. For example, RUN_CREATED creates the initial state, STEP_STARTED updates the current step, and WRITE_PERMISSION_REVOKED sets write_allowed to False.

The event sequence must be contiguous because each event builds on the state produced by the previous event. A gap means the intermediate state is unknown, so the sequence must be rejected.

After the run reaches a terminal state (COMPLETED or FAILED), no new events can be accepted because the state is final; applying further events would cause inconsistency.

Boundary decision for this stage: When the event sequence is not contiguous or the run is terminal, an exception must be raised rather than attempting to repair or ignore the problem. This ensures reliable and auditable state reconstruction.

Experiment: state-memory-context-01-state-and-events
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/state-memory-context/01-state-and-events/
Gate: Pending acceptance

Worked example

Goal: Demonstrate how reduce_events rebuilds RunState from three events and verify state fields.

Suppose we have a run: first the run is created (sequence 1), then a step starts (sequence 2), and finally write permission is revoked (sequence 3).

We call reduce_events with these three events; it will apply each event in order using apply_event to update the state.

The final state should be: status RUNNING, current_step “inspect”, write_allowed False, and version 3.

Implementation steps

Define three events: RUN_CREATED, STEP_STARTED, WRITE_PERMISSION_REVOKED.
Call reduce_events(events).
Assert that the returned state fields match expectations.

Verified solution code: forge/state/__init__.py

from .models import EventKind, RunEvent, RunState, RunStatus, apply_event, reduce_events

__all__ = ["EventKind", "RunEvent", "RunState", "RunStatus", "apply_event", "reduce_events"]

Verified solution code: forge/state/models.py

from __future__ import annotations

from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field

class RunStatus(StrEnum):
    CREATED = "created"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

class EventKind(StrEnum):
    RUN_CREATED = "run_created"
    STEP_STARTED = "step_started"
    WRITE_PERMISSION_REVOKED = "write_permission_revoked"
    RUN_COMPLETED = "run_completed"
    RUN_FAILED = "run_failed"

class RunEvent(BaseModel):
    model_config = ConfigDict(frozen=True)
    event_id: str = Field(min_length=1)
    run_id: str = Field(min_length=1)
    sequence: int = Field(ge=1)
    kind: EventKind
    payload: dict[str, Any] = Field(default_factory=dict)

class RunState(BaseModel):
    model_config = ConfigDict(frozen=True)
    run_id: str
    status: RunStatus
    version: int
    last_event_id: str
    current_step: str | None = None
    write_allowed: bool = True
    failure_reason: str | None = None

def apply_event(state: RunState | None, event: RunEvent) -> RunState:
    if state is None:
        if event.kind != EventKind.RUN_CREATED or event.sequence != 1:
            raise ValueError("the first event must create the run at sequence 1")
        return RunState(run_id=event.run_id, status=RunStatus.CREATED, version=1, last_event_id=event.event_id, write_allowed=bool(event.payload.get("write_allowed", True)))
    if event.run_id != state.run_id:
        raise ValueError("an event cannot be applied to a different run")
    if event.sequence != state.version + 1:
        raise ValueError("event sequence must be contiguous")
    if state.status in {RunStatus.COMPLETED, RunStatus.FAILED}:
        raise ValueError("a terminal run cannot accept more events")
    update: dict[str, Any] = {"version": event.sequence, "last_event_id": event.event_id}
    if event.kind == EventKind.STEP_STARTED:
        step = str(event.payload.get("step", "")).strip()
        if not step:
            raise ValueError("step_started requires a non-empty step")
        update.update(status=RunStatus.RUNNING, current_step=step)
    elif event.kind == EventKind.WRITE_PERMISSION_REVOKED:
        update["write_allowed"] = False
    elif event.kind == EventKind.RUN_COMPLETED:
        update.update(status=RunStatus.COMPLETED, current_step=None)
    elif event.kind == EventKind.RUN_FAILED:
        reason = str(event.payload.get("reason", "")).strip()
        if not reason:
            raise ValueError("run_failed requires a reason")
        update.update(status=RunStatus.FAILED, failure_reason=reason, current_step=None)
    elif event.kind == EventKind.RUN_CREATED:
        raise ValueError("a run can only be created once")
    return state.model_copy(update=update)

def reduce_events(events: list[RunEvent], initial: RunState | None = None) -> RunState:
    state = initial
    for event in events:
        state = apply_event(state, event)
    if state is None:
        raise ValueError("at least one event is required")
    return state

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Why this implementation

RUN_CREATED creates the initial state, setting run_id and write_allowed (default True).

STEP_STARTED updates the state to RUNNING and sets current_step to “inspect”.

WRITE_PERMISSION_REVOKED sets write_allowed to False.

The final state version is 3 because the last event’s sequence is 3.

Guided practice

Complete the missing sequence continuity check in apply_event to pass test_sequence_gaps_are_rejected.

Open forge/state/models.py and locate the apply_event function.
After the run_id check and before the terminal state check, add a sequence continuity check: if event.sequence != state.version + 1, raise ValueError.
Run the test test_sequence_gaps_are_rejected and confirm it passes.

Progressive hints

Hint 1: The state object has a version field representing the latest applied event sequence number.
Hint 2: The new event’s sequence must be exactly one greater than the current version.
Hint 3: The error message should contain “contiguous” to satisfy the test assertion.

Success criteria

test_sequence_gaps_are_rejected passes
apply_event raises ValueError when the sequence is not contiguous

Fault injection and diagnosis

The failure fixture at fixtures/failure.json simulates an event log that skips sequence 2. Running test_sequence_gaps_are_rejected fails because reduce_events accepts sequence 1 followed directly by sequence 3 instead of raising ValueError. You need to diagnose and fix this issue.

In this stage, you have learned the principle that event sequences must be contiguous and that apply_event is responsible for applying events. The defect in the failure fixture is the missing continuity check, which is exactly what you completed in guidedPractice. By inspecting apply_event, you can identify the missing line if event.sequence != state.version + 1 and restore it.

Verified fault contract

Target test: test_sequence_gaps_are_rejected
Faulty file: forge/state/models.py

Diagnosis order

Run the faulty command: python -m pytest -q ../tests/test_stage.py::test_sequence_gaps_are_rejected
Observe the output: the test fails with AssertionError: a sequence gap must be rejected
Inspect apply_event in forge/state/models.py and notice the missing sequence continuity check.
Confirm that after the run_id check there is no line if event.sequence != state.version + 1.

Fix principle: In apply_event, after the run_id check and before the terminal state check, add: if event.sequence != state.version + 1: raise ValueError(“event sequence must be contiguous”).

python -m pytest -q ../tests/test_stage.py::test_sequence_gaps_are_rejected

Transfer challenge

Office research scenario: a run is created (sequence 1), then document retrieval starts (sequence 2), then write permission is revoked (sequence 3), and finally a brief is generated (sequence 4).

Changed condition: The content and order of the event sequence differ from the original example, but the event types and constraints remain the same.

Use reduce_events to rebuild state from the above event sequence and verify that write_allowed is False, current_step is “generate_brief”, and status is RUNNING.

Success criteria

The state object’s write_allowed attribute is False
The state object’s current_step attribute is “generate_brief”
The state object’s status attribute is RunStatus.RUNNING

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Design explanation question: Why must the event sequence be contiguous? Why can’t a completed run accept more events?

A strong answer should include

The event sequence must be contiguous because each event builds on the state produced by the previous event. A gap means the intermediate state is unknown, making state reconstruction unreliable.
Once a run is completed, its state is final. Applying further events would lead to inconsistency, such as a STEP_STARTED after COMPLETED, which violates business rules.

Mastery evidence

All three tests pass: test_events_rebuild_the_same_state_deterministically, test_sequence_gaps_are_rejected, test_terminal_runs_reject_later_events
Can explain the difference between state and events: state is a projection of current progress, events are immutable historical facts.
Can explain why event sequences must be contiguous and why terminal states cannot accept new events.

STAGE 02

Distinguish a session from an individual run and explicitly manage start, finish, and concurrency boundaries

Start with a concrete problem

Why do we need to distinguish sessions from runs?

In the previous stage, we built state and event models, but we still lack the ability to manage multiple runs. Now, a single user session may contain several runs, such as a retry after a failure or parallel processing of multiple tasks. If every run uses the same run_id, histories become mixed, and we cannot tell which run produced which events.

This stage requires implementing a Session container that can start multiple runs, finish runs, and guarantee that run_id is unique within a session. The test test_session_tracks_multiple_runs_without_merging_them verifies that the session correctly tracks multiple runs without merging them; the test test_duplicate_run_id_is_rejected verifies that a duplicate run_id is rejected.

Make a prediction

When implementing start_run, what should happen if an already existing run_id is passed?

Silently ignore and do not add a new run
Raise ValueError because run_id must be unique
Overwrite the old run record
Automatically generate a new run_id

Reasoning guide: The correct answer is to raise ValueError. The test test_duplicate_run_id_is_rejected explicitly requires that a duplicate run_id be rejected, and the error message must contain ‘unique’. Silently ignoring or overwriting would cause history loss or confusion, and automatically generating a new ID would violate the caller’s intent.

What it is

A session is a container for runs

Session is an immutable data structure containing a session_id and a tuple of runs. Each run is a RunRecord with a run_id and a status. The start_run function adds a new RunRecord to the session and returns a new Session instance. The finish_run function changes the status of a specified run to a terminal state (completed, failed, or cancelled).

The session lifecycle is explicitly managed by these functions: create session, start run, finish run. Run IDs must be unique within a session, which is guaranteed by a check in start_run.

What it is not

A session is not a single run

A session is not one concrete execution but a container for multiple executions. A run is a single execution with explicit state transitions. Confusing the two leads to state management chaos, such as being unable to distinguish events from different runs.

A session is also not equivalent to chat history or context; it only stores structural information about runs, not the actual event content.

How it relates to neighboring concepts

Relationship among session, runs, and state

Session contains multiple RunRecords, each with a run_id and status. start_run and finish_run are functions that operate on Session and return new Session instances, preserving immutability. The active_run_ids property filters runs with status ‘active’ and returns their IDs.

This design allows session state to be safely passed and shared without accidental modification.

Boundary decision for this stage: In start_run, we must check whether the run_id already exists; if it does, raise ValueError. This is the boundary for run ID uniqueness within a session.

Experiment: state-memory-context-02-session-lifecycle
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/state-memory-context/02-session-lifecycle/
Gate: Pending acceptance

Worked example

Goal: Demonstrate creating a session, starting two runs, finishing one, and verifying the active run list.

Suppose we have a session session-a and need to start run-1 and run-2 sequentially, then finish run-1. Finally, the active runs should be only run-2, and the runs list should contain run-1 (completed) and run-2 (active) in start order.

This example shows the basic usage of Session and how start_run and finish_run cooperate to manage run states.

Implementation steps

Create a Session instance: session = Session(session_id=“session-a”)
Start the first run: session = start_run(session, “run-1”)
Start the second run: session = start_run(session, “run-2”)
Finish the first run: session = finish_run(session, “run-1”, “completed”)
Verify active_run_ids is (“run-2”,) and runs list order is correct.

Verified solution code: forge/state/session.py

from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field

class RunRecord(BaseModel):
    model_config = ConfigDict(frozen=True)
    run_id: str
    status: str = "active"

class Session(BaseModel):
    model_config = ConfigDict(frozen=True)
    session_id: str
    runs: tuple[RunRecord, ...] = Field(default_factory=tuple)
    @property
    def active_run_ids(self) -> tuple[str, ...]:
        return tuple(run.run_id for run in self.runs if run.status == "active")

def start_run(session: Session, run_id: str) -> Session:
    if any(run.run_id == run_id for run in session.runs):
        raise ValueError("run_id must be unique inside a session")
    return session.model_copy(update={"runs": (*session.runs, RunRecord(run_id=run_id))})

def finish_run(session: Session, run_id: str, status: str) -> Session:
    if status not in {"completed", "failed", "cancelled"}:
        raise ValueError("finish status must be terminal")
    found = False
    updated = []
    for run in session.runs:
        if run.run_id == run_id:
            found = True
            if run.status != "active":
                raise ValueError("run is already terminal")
            updated.append(run.model_copy(update={"status": status}))
        else:
            updated.append(run)
    if not found:
        raise KeyError(run_id)
    return session.model_copy(update={"runs": tuple(updated)})

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Why this implementation

start_run adds a new RunRecord each time, with status defaulting to ‘active’. finish_run finds the matching run_id and changes its status to ‘completed’. Since Session is immutable, each operation returns a new instance.

The active_run_ids property uses a list comprehension to filter run_ids with status ‘active’, so it ends up containing only run-2.

Guided practice

Add a duplicate run_id check in start_run to pass test_duplicate_run_id_is_rejected.

Open the file forge/state/session.py.
In the start_run function, before adding a new RunRecord, check whether session.runs already contains a record with the same run_id.
If it exists, raise ValueError with a message containing ‘unique’.
Run the tests to verify.

Progressive hints

Use the any() function to iterate over session.runs and compare run.run_id with the passed run_id.
The error message can be “run_id must be unique inside a session”.

Success criteria

test_duplicate_run_id_is_rejected passes.
test_session_tracks_multiple_runs_without_merging_them still passes.

Fault injection and diagnosis

In the fault fixture fixtures/failure.json, the start_run function in forge/state/session.py does not include a duplicate run_id check. Running the test test_duplicate_run_id_is_rejected fails because when start_run is called a second time with the same run_id, the duplicate is appended to the session, causing the test to raise AssertionError.

The core of this stage is implementing the run_id uniqueness check. The learner has already added this check in forge/state/session.py during guidedPractice, so they can identify the fault cause and fix it. The start_run function in the fault fixture is identical to the version before guidedPractice, missing the check.

Verified fault contract

Target test: test_duplicate_run_id_is_rejected
Faulty file: forge/state/session.py

Diagnosis order

Run the faulty test command and observe the AssertionError.
Inspect the start_run function in forge/state/session.py and notice there is no duplicate run_id check.
Compare with the correct implementation and confirm the missing any() check.

Fix principle: In forge/state/session.py, add a duplicate check in start_run: if any(run.run_id == run_id for run in session.runs): raise ValueError(“run_id must be unique inside a session”).

python -m pytest -q ../tests/test_stage.py::test_duplicate_run_id_is_rejected

Transfer challenge

In a music creation scenario, use Session to manage multiple track runs. Each track run represents one creation attempt, such as generating a melody or arrangement. You need to ensure that track run IDs are unique within the session, for example, there cannot be two ‘track-1’ runs.

Changed condition: The scenario changes from software delivery to music creation, but the Session and run mechanism remain unchanged.

Write code that uses Session to start multiple track runs, and deliberately attempt to start a duplicate track run ID, verifying that a ValueError is raised.

Success criteria

Able to create a Session and start multiple runs with different IDs.
Attempting to start a duplicate ID raises ValueError.
The active run list is correct.

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

..                                                                       [100%]
2 passed in <duration>s

Design explanation question: Why must run IDs be unique within a session? Based on the start_run check mechanism implemented in this stage, explain what specific problems would arise if duplicate IDs were allowed.

A strong answer should include

Run IDs identify a single execution; if duplicated, it is impossible to distinguish events and states of different runs.
Uniqueness ensures that histories are not mixed, for example, retries do not overwrite or merge previous runs.
The any() check in start_run verifies that the ID does not exist before adding a new run, thereby maintaining the distinguishability of runs within a session.

Mastery evidence

Both tests pass.
Able to explain the difference between session and run.
Able to implement duplicate ID check in start_run.

STAGE 03

Persist a minimal versioned checkpoint with integrity verification and reject corrupt or unsupported snapshots

Start with a concrete problem

Saved checkpoints may be tampered with or version-incompatible

During long-running tasks, we need to save the current state to pause, resume, or replay. If a checkpoint file is accidentally modified (e.g., disk error or deliberate tampering), recovery may load an incorrect state, causing subsequent decisions to be based on untrusted data.

Additionally, as the system evolves, the checkpoint structure may change. Loading an old or unknown version may crash due to missing fields or type mismatches, or silently produce wrong results.

This stage requires implementing a minimal checkpoint module with a version number and SHA-256 checksum, verifying both on load and rejecting corrupt or unsupported snapshots.

Make a prediction

When implementing checkpoint loading, which approach most reliably prevents loading tampered state?

Only check that the JSON file can be parsed into a CheckpointEnvelope model.
Recompute the SHA-256 checksum of the state on load and compare it with the stored checksum.
Trust file system permissions, assuming only authorized users can modify the file.
Only check that schema_version equals the current version on load.

Reasoning guide: The correct option is to recompute the checksum and compare. Parsing JSON or checking the version alone cannot detect content tampering; file system permissions cannot prevent all accidental modifications; version checking only handles structural compatibility, not data integrity.

What it is

Checkpoint Contract

A checkpoint is an immutable envelope (CheckpointEnvelope) containing schema_version, state, and checksum_sha256. schema_version identifies the structure version, state is the current run state, and checksum_sha256 is the SHA-256 hash of the canonical JSON representation of schema_version and state.

On save, we compute the checksum and atomically write the file; on load, we first verify that schema_version is supported, then recompute the checksum and compare it with the stored value, raising ValueError on any mismatch.

What it is not

What a Checkpoint Is Not

A checkpoint is not a simple serialization of state; it must include an integrity proof. It is also not a substitute for long-term memory or session history; it only saves a snapshot of state at a point in time.

A checkpoint does not decide when to save or restore, nor does it handle concurrent write conflicts; those policies are determined by upper-level session management.

How it relates to neighboring concepts

Relationships with Other Concepts

A checkpoint is closely related to state (RunState): it encapsulates the state and provides persistence. Unlike context, a checkpoint does not automatically enter the model context; only after explicit loading and selection of part of the state does it become part of the context.

In terms of observability, the checkpoint’s version and checksum provide an audit trail, helping diagnose recovery failures.

Boundary decision for this stage: When loading a checkpoint, both schema_version and checksum_sha256 must be verified; any failure should reject recovery rather than attempt partial loading or silent repair.

Experiment: state-memory-context-03-checkpoint-contract
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/state-memory-context/03-checkpoint-contract/
Gate: Pending acceptance

Worked example

Goal: Demonstrate saving and loading a checkpoint and verifying exact state preservation.

We start with a simple run state: first create a run, then start a step, and finally revoke write permission. We obtain a RunState via reduce_events.

Use save_checkpoint to save the state to a file in a temporary directory, returning a CheckpointEnvelope. Then use load_checkpoint to load that file and assert that the loaded state equals the saved state and the checksum length is 64.

Implementation steps

Construct three events: RUN_CREATED (write_allowed=True), STEP_STARTED (step=‘inspect’), WRITE_PERMISSION_REVOKED.
Call reduce_events to get state.
Call save_checkpoint(path, state), where path is tmp_path / ‘run-checkpoint.json’.
Call load_checkpoint(path) to get loaded.
Assert loaded == saved, loaded.state == state, loaded.state.write_allowed is False, len(loaded.checksum_sha256) == 64.

Verified solution code: forge/state/checkpoint.py

from __future__ import annotations

import hashlib
import hmac
import json
import os
from pathlib import Path
from typing import Any

from pydantic import BaseModel, ConfigDict, Field

from .models import RunState

CHECKPOINT_SCHEMA_VERSION = 1


class CheckpointEnvelope(BaseModel):
    model_config = ConfigDict(frozen=True)

    schema_version: int = Field(ge=1)
    state: RunState
    checksum_sha256: str = Field(min_length=64, max_length=64)


def _canonical_payload(schema_version: int, state: RunState) -> bytes:
    payload: dict[str, Any] = {
        "schema_version": schema_version,
        "state": state.model_dump(mode="json"),
    }
    return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")


def checkpoint_checksum(schema_version: int, state: RunState) -> str:
    return hashlib.sha256(_canonical_payload(schema_version, state)).hexdigest()


def save_checkpoint(path: str | Path, state: RunState) -> CheckpointEnvelope:
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    envelope = CheckpointEnvelope(
        schema_version=CHECKPOINT_SCHEMA_VERSION,
        state=state,
        checksum_sha256=checkpoint_checksum(CHECKPOINT_SCHEMA_VERSION, state),
    )
    temporary = target.with_name(f".{target.name}.tmp")
    temporary.write_text(
        json.dumps(envelope.model_dump(mode="json"), ensure_ascii=False, sort_keys=True, indent=2) + "\n",
        encoding="utf-8",
    )
    os.replace(temporary, target)
    return envelope


def load_checkpoint(path: str | Path) -> CheckpointEnvelope:
    raw = json.loads(Path(path).read_text(encoding="utf-8"))
    envelope = CheckpointEnvelope.model_validate(raw)
    if envelope.schema_version != CHECKPOINT_SCHEMA_VERSION:
        raise ValueError(f"unsupported checkpoint schema version: {envelope.schema_version}")
    expected = checkpoint_checksum(envelope.schema_version, envelope.state)
    if not hmac.compare_digest(expected, envelope.checksum_sha256):
        raise ValueError("checkpoint checksum mismatch")
    return envelope

Verified solution code: forge/state/__init__.py

from .checkpoint import CheckpointEnvelope, checkpoint_checksum, load_checkpoint, save_checkpoint
from .models import EventKind, RunEvent, RunState, RunStatus, apply_event, reduce_events
from .session import RunRecord, Session, finish_run, start_run

__all__ = [
    "CheckpointEnvelope",
    "EventKind",
    "RunEvent",
    "RunRecord",
    "RunState",
    "RunStatus",
    "Session",
    "apply_event",
    "checkpoint_checksum",
    "finish_run",
    "load_checkpoint",
    "reduce_events",
    "save_checkpoint",
    "start_run",
]

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Why this implementation

save_checkpoint internally creates a CheckpointEnvelope, computes the SHA-256 checksum of the canonical payload, and atomically writes the file.

load_checkpoint reads the JSON, validates schema_version and checksum, then returns the envelope. Since the file is unmodified, the checksum matches and loading succeeds.

Guided practice

Restore checksum comparison in load_checkpoint to pass test_checkpoint_corruption_is_rejected.

Open forge/state/checkpoint.py and locate the load_checkpoint function.
After the version check, call checkpoint_checksum(envelope.schema_version, envelope.state) to compute the expected checksum.
Use hmac.compare_digest(expected, envelope.checksum_sha256) to compare the expected value with the stored value.
If they do not match, raise ValueError(“checkpoint checksum mismatch”).
Run the tests to verify.

Progressive hints

The checkpoint_checksum function already exists; it accepts schema_version and state and returns a hex string.
hmac.compare_digest is used for constant-time comparison to avoid timing attacks.
Ensure the check is performed before returning the envelope.

Success criteria

test_checkpoint_corruption_is_rejected passes.
load_checkpoint raises ValueError containing ‘checksum mismatch’ after tampering.

Fault injection and diagnosis

The failure fixture fixtures/failure.json describes a checkpoint corruption scenario: the state is manually edited without recomputing the SHA-256 digest, so recovery must stop. The current faulty code skips the checksum comparison in load_checkpoint, causing test_checkpoint_corruption_is_rejected to fail. The executable faulty fixture modifies forge/state/checkpoint.py and is reliably reproduced by test_checkpoint_corruption_is_rejected; the verified root cause is: load_checkpoint validates the JSON shape but no longer compares the stored SHA-256 digest with the canonical state payload.

This stage’s core is implementing integrity verification. The learner has already added the checksum comparison in load_checkpoint through guidedPractice, so they can recognize the missing check in the faulty code and know how to fix it.

Verified fault contract

Target test: test_checkpoint_corruption_is_rejected
Faulty file: forge/state/checkpoint.py

Diagnosis order

Run the failing test: python -m pytest -q ../tests/test_stage.py::test_checkpoint_corruption_is_rejected, observe the failure message ‘Failed: DID NOT RAISE ValueError’.
Inspect the load_checkpoint function and notice that there is only a schema_version check, no checksum comparison.
Compare with the correct implementation and confirm the missing expected = checkpoint_checksum(…) and hmac.compare_digest check.

Fix principle: In load_checkpoint, after the version check, add: expected = checkpoint_checksum(envelope.schema_version, envelope.state); if not hmac.compare_digest(expected, envelope.checksum_sha256): raise ValueError(“checkpoint checksum mismatch”).

python -m pytest -q ../tests/test_stage.py::test_checkpoint_corruption_is_rejected

Transfer challenge

In a software delivery scenario, you are adding a feature to a frontend repository. The run state includes the current branch, list of modified files, and test results. You need to save a checkpoint, then simulate file tampering and verify that loading fails.

Changed condition: The scenario data changes from a generic run state to a specific software delivery state, but the checkpoint mechanism remains unchanged.

Using the existing save_checkpoint and load_checkpoint, create a checkpoint containing software delivery state, manually modify a field in the JSON file (e.g., change write_allowed to True), then attempt to load and confirm that a ValueError is raised.

Success criteria

Successfully save a checkpoint and load the unmodified file.
After modifying the file, loading raises ValueError containing ‘checksum mismatch’.

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Design explanation question: Why use SHA-256 instead of a simple hash (like CRC32) to protect checkpoint integrity?

A strong answer should include

SHA-256 is a cryptographic hash with collision resistance, making it difficult to find two different inputs producing the same hash.
CRC32 is designed for detecting random errors, not tamper resistance; an attacker can easily construct collisions.
Using SHA-256 ensures that even if an attacker knows the algorithm, they cannot modify the state without changing the hash.

Mastery evidence

All three tests pass: test_checkpoint_round_trip_preserves_exact_state, test_checkpoint_corruption_is_rejected, test_unknown_checkpoint_version_is_rejected.
Can explain the roles of version number and checksum in a checkpoint and why both are needed.
Successfully diagnosed and fixed the missing checksum comparison in the fault lab.

STAGE 04

Recover from a checkpoint plus tail events and prove that the result matches a full event replay

Start with a concrete problem

How can we prove that state recovered from a checkpoint matches a full event replay exactly?

During long-running tasks, the system periodically saves checkpoints to support pause and resume. When a task is interrupted, we want to restore the run state by loading a checkpoint and replaying the events that occurred after it (the tail events).

However, simply loading a checkpoint and applying tail events does not guarantee correct recovery. If the tail events are incomplete (for example, one event is missing), the recovered state may differ from the state obtained by replaying all events from the beginning.

In this stage, you will implement a verification function, verify_recovery_equivalent, that compares the state recovered from checkpoint plus tail events with the state from a full event replay, and raises RecoveryMismatch if they differ.

Make a prediction

Before implementing recovery equivalence verification, predict: what happens if a tail event is omitted during recovery from a checkpoint?

The recovered state may differ from full replay, but the system cannot automatically detect it
The recovered state will always match full replay because the checkpoint contains all necessary information
The system will raise an exception because the checkpoint checksum detects the missing event
The recovered state may differ, but it can be manually checked via logs

Reasoning guide: The correct answer is the first option. A checkpoint only stores a snapshot of state at a certain point; it cannot know which events should occur afterwards. If tail events are incomplete, the recovered state will miss the effects of those events, causing a mismatch with full replay. Therefore, an explicit comparison is needed to detect this inconsistency.

What it is

What recovery equivalence verification is

Recovery equivalence verification is a safety mechanism that compares the results of two computation paths to ensure correct recovery: path one loads state from a checkpoint and applies tail events; path two replays the full event history from the beginning.

If the states from both paths are exactly equal, it proves that the checkpoint and tail events are sufficient to reconstruct the full state; if they differ, it indicates missing or out-of-order tail events, and an exception must be raised.

What it is not

What recovery equivalence verification is not

It is not checkpoint checksum verification. A checksum only ensures the checkpoint file has not been tampered with; it cannot guarantee tail event completeness.

It is not a superficial state comparison; it must actually perform a full replay because state may contain complex objects and surface-level field checks cannot determine deep consistency.

How it relates to neighboring concepts

Relationship with other concepts

A checkpoint saves a state snapshot at a certain moment and serves as the starting point for recovery.

Event replay builds state by applying events in sequence and represents the complete path.

Recovery equivalence verification combines both: the checkpoint provides acceleration, tail events provide increments, and full replay provides the baseline.

Boundary decision for this stage: Recovery is considered equivalent if and only if the state recovered from checkpoint plus tail events is exactly equal to the state from full event replay; otherwise, a RecoveryMismatch exception must be raised to prevent incorrect state from propagating.

Experiment: state-memory-context-04-recovery-equivalence
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/state-memory-context/04-recovery-equivalence/
Gate: Pending acceptance

Worked example

Goal: Demonstrate how to recover from a checkpoint plus tail events and compare with full replay to ensure consistency.

Suppose a run history contains four events: RUN_CREATED, STEP_STARTED(inspect), WRITE_PERMISSION_REVOKED, STEP_STARTED(report). We save a checkpoint after the second event.

Now the task is interrupted. We load the checkpoint and replay the tail events (the third and fourth events). The recovered state should be exactly the same as the state obtained by replaying all four events from the beginning.

If the tail events omit the fourth event, the recovered state will stop after the third event and differ from full replay; in that case, the verification function should raise RecoveryMismatch.

Implementation steps

Use save_checkpoint to save a checkpoint after the second event.
Call recover_state to load the checkpoint and apply tail events, obtaining the recovered state.
Call reduce_events to replay the full event history, obtaining the baseline state.
Compare the two states; if equal, return the recovered state, otherwise raise RecoveryMismatch.

Verified solution code: forge/state/recovery.py

from __future__ import annotations

from pathlib import Path

from .checkpoint import load_checkpoint
from .models import RunEvent, RunState, reduce_events


class RecoveryMismatch(RuntimeError):
    pass


def recover_state(checkpoint_path: str | Path, tail_events: list[RunEvent]) -> RunState:
    checkpoint = load_checkpoint(checkpoint_path)
    if tail_events and tail_events[0].sequence != checkpoint.state.version + 1:
        raise ValueError("tail events must continue immediately after the checkpoint")
    if not tail_events:
        return checkpoint.state
    return reduce_events(tail_events, initial=checkpoint.state)


def verify_recovery_equivalent(
    checkpoint_path: str | Path,
    tail_events: list[RunEvent],
    full_history: list[RunEvent],
) -> RunState:
    recovered = recover_state(checkpoint_path, tail_events)
    replayed = reduce_events(full_history)
    if recovered != replayed:
        raise RecoveryMismatch(
            "checkpoint recovery differs from full event replay; keep the original checkpoint and inspect the tail boundary"
        )
    return recovered

Verified solution code: forge/state/__init__.py

from .checkpoint import CheckpointEnvelope, checkpoint_checksum, load_checkpoint, save_checkpoint
from .models import EventKind, RunEvent, RunState, RunStatus, apply_event, reduce_events
from .recovery import RecoveryMismatch, recover_state, verify_recovery_equivalent
from .session import RunRecord, Session, finish_run, start_run

__all__ = [
    "CheckpointEnvelope",
    "EventKind",
    "RecoveryMismatch",
    "RunEvent",
    "RunRecord",
    "RunState",
    "RunStatus",
    "Session",
    "apply_event",
    "checkpoint_checksum",
    "finish_run",
    "load_checkpoint",
    "recover_state",
    "reduce_events",
    "save_checkpoint",
    "start_run",
    "verify_recovery_equivalent",
]

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Why this implementation

This test verifies that when tail events are complete, the recovered state matches full replay.

verify_recovery_equivalent internally first calls recover_state to get the recovered state, then calls reduce_events to get the replayed state, and then compares them.

If equal, it returns the recovered state; if not, it raises RecoveryMismatch.

Guided practice

Add full replay comparison in verify_recovery_equivalent to pass test_missing_tail_event_is_detected_by_equivalence_check.

Open the file forge/state/recovery.py.
In the verify_recovery_equivalent function, call recover_state to obtain the recovered state.
Call reduce_events(full_history) to obtain the full replay state.
Compare the two states; if they are not equal, raise RecoveryMismatch with a message containing ‘differs from full event replay’.
Run the tests to confirm they pass.

Progressive hints

Use the != operator to compare two RunState objects.
The exception message must contain ‘differs from full event replay’ to match the test.
Ensure the replayed state is calculated before the comparison.

Success criteria

test_missing_tail_event_is_detected_by_equivalence_check passes
test_checkpoint_plus_tail_matches_full_replay still passes
test_tail_must_start_at_checkpoint_boundary still passes

Fault injection and diagnosis

In the current faulty version of recovery.py, the verify_recovery_equivalent function only returns the result of recover_state without comparing it to a full replay. Therefore, when a tail event is missing, the test test_missing_tail_event_is_detected_by_equivalence_check expects RecoveryMismatch to be raised, but it is not, causing the test to fail. The executable faulty fixture modifies forge/state/recovery.py and is reliably reproduced by test_missing_tail_event_is_detected_by_equivalence_check; the verified root cause is: verify_recovery_equivalent returns recovered state without calculating and comparing replayed full_history.

This stage has taught the concept of recovery equivalence and the necessity of comparing checkpoint plus tail events with full replay. The faulty version lacks this comparison step, so we can fix it by adding the comparison logic.

Verified fault contract

Target test: test_missing_tail_event_is_detected_by_equivalence_check
Faulty file: forge/state/recovery.py

Diagnosis order

Run the test test_missing_tail_event_is_detected_by_equivalence_check and observe the failure message: DID NOT RAISE RecoveryMismatch.
Open forge/state/recovery.py and inspect the verify_recovery_equivalent function.
Notice that the function calls recover_state and immediately returns the result without computing replayed = reduce_events(full_history) and without comparing recovered and replayed.
Confirm that the missing comparison is the root cause of the test failure.

Fix principle: In verify_recovery_equivalent, compute replayed = reduce_events(full_history), then compare recovered and replayed; if they are not equal, raise RecoveryMismatch. This fix ensures that any missing or out-of-order tail events are detected, preventing incorrect state from being accepted as a correct recovery result.

python -m pytest -q ../tests/test_stage.py::test_missing_tail_event_is_detected_by_equivalence_check

Transfer challenge

Office research scenario: You are working on a decision brief. The event history includes: run created, start research step, revoke write permission, start report step. You save a checkpoint after the second event. Now you need to recover from the checkpoint, but deliberately omit the last event (start report step), then call verify_recovery_equivalent to verify equivalence.

Changed condition: Scenario data changes from software delivery to office research; event content differs but structure is the same.

Using the office research event history, construct the events with RunEvent and EventKind. Save a checkpoint after the second event using save_checkpoint. Then call verify_recovery_equivalent with the checkpoint path, a shortened tail containing only the third event, and the full event history. Assert that RecoveryMismatch is raised.

Success criteria

Correctly construct the office research event history.
Save a checkpoint after the second event.
Call verify_recovery_equivalent with the shortened tail events.
Assert that RecoveryMismatch is raised.

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Design explanation question: Why must tail events start immediately after the checkpoint boundary?

A strong answer should include

The checkpoint stores a state at a certain version, and its version field indicates the sequence number of the last applied event.
Tail events must start from version+1; otherwise, events would be skipped or applied twice, leading to inconsistent state.
The recover_state function enforces this by checking that the first tail event’s sequence equals checkpoint.state.version + 1, otherwise raising ValueError.

Mastery evidence

All three tests pass: test_checkpoint_plus_tail_matches_full_replay, test_tail_must_start_at_checkpoint_boundary, test_missing_tail_event_is_detected_by_equivalence_check.
Can explain the principle of recovery equivalence: comparing checkpoint plus tail events with full replay.
Can diagnose and fix the fault of missing full replay comparison.
Can apply recovery equivalence verification in different scenarios.

STAGE 05

Separate recoverable runtime state from retrievable long-term memory with scope, expiry, and deletion semantics

Start with a concrete problem

Separate long-term memory from runtime state and ensure memory is retrievable only within correct scope and validity period

In the previous stage, we implemented recoverable runtime state but no long-term memory. Now we need to add forge/memory.py, define MemoryScope (RUN, SESSION, USER) and MemoryRecord, and implement MemoryStore methods put, query, delete, and purge_expired.

Key constraints: queries must filter by scope and subject and exclude expired memories; deletion must be explicit and idempotent. Tests test_memory_is_queried_only_inside_its_scope_and_subject, test_expired_memory_is_hidden_and_can_be_purged, and test_deletion_is_explicit_and_idempotent will verify these behaviors.

The current starting code has no forge/memory.py, so running the tests will raise ModuleNotFoundError. You need to create that file from scratch and modify forge/init.py to export the relevant symbols.

Make a prediction

When implementing MemoryStore.query, what happens if you filter only by scope and subject_id without checking expires_at?

Expired memories will still be returned, causing later decisions to use stale information.
The query will raise an exception because the SQL statement is incomplete.
Expired memories will be automatically deleted by the database, so they won’t be returned.
The query results will be sorted by creation time but will not include expired memories.

Reasoning guide: The correct option is the first one. If the query does not check expires_at, then even if a memory is expired, as long as scope and subject_id match, it will be returned. This causes expired memories to contaminate the current context and affect decisions. The test test_expired_memory_is_hidden_and_can_be_purged specifically verifies this.

What it is

Memory boundary model

Memory is a data record with a scope and a lifetime. The scope (RUN, SESSION, USER) defines the visibility range of the memory: RUN is visible only within a single run, SESSION within a session, and USER across sessions long-term.

Each memory has created_at and optional expires_at. When querying, the system filters out expired memories based on the current time. Deletion is performed by memory_id precisely, and repeated deletion does not raise an error; it returns False to indicate no record was deleted.

What it is not

What memory boundaries are not

Memory is not a simple compressed chat transcript, nor does it put all historical information into context. It must define write, read, conflict, retention, and forgetting policies.

Memory is not the same as runtime state. Runtime state (such as current step, retry count) is volatile, while memory is persistent and can be reused across runs.

How it relates to neighboring concepts

Relationship between memory and other concepts

Memory and context: Context is the information visible to one model call, and memory is one possible source of context. Only memories that are queried and placed into context affect the current call.

Memory and state: State represents current progress, while memory preserves information across steps or sessions. State can be restored via checkpoints, while memory manages its lifecycle through scope and expiry policies.

Boundary decision for this stage: When implementing MemoryStore.query, you must filter by scope, subject_id, and expires_at simultaneously to ensure only currently valid memories within the requested scope are returned.

Experiment: state-memory-context-05-memory-boundaries
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/state-memory-context/05-memory-boundaries/
Gate: Pending acceptance

Worked example

Goal: Demonstrate creating a MemoryStore, inserting memories with different scopes, and querying a specific scope.

We first create the forge/memory.py file, defining the MemoryScope enum and MemoryRecord model. MemoryScope contains RUN, SESSION, and USER values. MemoryRecord contains memory_id, scope, subject_id, content, created_at, and optional expires_at.

Then we implement the MemoryStore class. The constructor takes a database path, creates a SQLite connection and table. The put method inserts a record, query filters by scope and subject, delete removes by ID, and purge_expired deletes all expired records.

In the worked example, we insert one RUN-scoped memory and one SESSION-scoped memory, then query the RUN scope to verify only the RUN memory is returned.

Implementation steps

Create forge/memory.py, define MemoryScope and MemoryRecord.
Implement MemoryStore.init, establish database connection and table schema.
Implement put method, insert record and handle uniqueness conflicts.
Implement query method, filter by scope and subject_id, and exclude expired records.
Implement delete and purge_expired methods.
Export MemoryRecord, MemoryScope, MemoryStore in forge/init.py.

Verified solution code: forge/memory.py

from __future__ import annotations

import sqlite3
from datetime import datetime, timezone
from enum import StrEnum
from pathlib import Path

from pydantic import BaseModel, ConfigDict, Field


class MemoryScope(StrEnum):
    RUN = "run"
    SESSION = "session"
    USER = "user"


class MemoryRecord(BaseModel):
    model_config = ConfigDict(frozen=True)

    memory_id: str = Field(min_length=1)
    scope: MemoryScope
    subject_id: str = Field(min_length=1)
    content: str = Field(min_length=1)
    created_at: datetime
    expires_at: datetime | None = None


class MemoryStore:
    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.connection = sqlite3.connect(self.path)
        self.connection.row_factory = sqlite3.Row
        self.connection.execute("PRAGMA journal_mode=WAL")
        self.connection.execute(
            """
            CREATE TABLE IF NOT EXISTS memories (
                memory_id TEXT PRIMARY KEY,
                scope TEXT NOT NULL,
                subject_id TEXT NOT NULL,
                content TEXT NOT NULL,
                created_at TEXT NOT NULL,
                expires_at TEXT
            )
            """
        )
        self.connection.execute(
            "CREATE INDEX IF NOT EXISTS idx_memories_scope_subject ON memories(scope, subject_id)"
        )
        self.connection.commit()

    def close(self) -> None:
        self.connection.close()

    def put(self, record: MemoryRecord) -> None:
        if record.expires_at is not None and record.expires_at <= record.created_at:
            raise ValueError("expires_at must be later than created_at")
        try:
            self.connection.execute(
                "INSERT INTO memories(memory_id, scope, subject_id, content, created_at, expires_at) VALUES (?, ?, ?, ?, ?, ?)",
                (
                    record.memory_id,
                    record.scope.value,
                    record.subject_id,
                    record.content,
                    record.created_at.isoformat(),
                    record.expires_at.isoformat() if record.expires_at else None,
                ),
            )
            self.connection.commit()
        except sqlite3.IntegrityError as error:
            raise ValueError("memory_id must be unique") from error

    def query(
        self,
        scope: MemoryScope,
        subject_id: str,
        *,
        now: datetime | None = None,
    ) -> list[MemoryRecord]:
        current = (now or datetime.now(timezone.utc)).isoformat()
        rows = self.connection.execute(
            """
            SELECT memory_id, scope, subject_id, content, created_at, expires_at
            FROM memories
            WHERE scope = ? AND subject_id = ?
              AND (expires_at IS NULL OR expires_at > ?)
            ORDER BY created_at, memory_id
            """,
            (scope.value, subject_id, current),
        ).fetchall()
        return [MemoryRecord.model_validate(dict(row)) for row in rows]

    def delete(self, memory_id: str) -> bool:
        cursor = self.connection.execute("DELETE FROM memories WHERE memory_id = ?", (memory_id,))
        self.connection.commit()
        return cursor.rowcount == 1

    def purge_expired(self, *, now: datetime | None = None) -> int:
        current = (now or datetime.now(timezone.utc)).isoformat()
        cursor = self.connection.execute(
            "DELETE FROM memories WHERE expires_at IS NOT NULL AND expires_at <= ?",
            (current,),
        )
        self.connection.commit()
        return cursor.rowcount

Verified solution code: forge/__init__.py

"""HeatStack Forge teaching package."""

from .memory import MemoryRecord, MemoryScope, MemoryStore

__all__ = ["MemoryRecord", "MemoryScope", "MemoryStore"]

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Why this implementation

MemoryStore.query uses a SQL WHERE clause to filter scope, subject_id, and expires_at simultaneously. expires_at being NULL means never expires; otherwise it must be greater than the current time.

In the test, RUN and SESSION memories are inserted, and querying the RUN scope returns only the RUN memory because the SQL filters by scope.

Guided practice

Restore expiry filtering in MemoryStore.query to pass test_expired_memory_is_hidden_and_can_be_purged.

Open forge/memory.py and locate the query method.
Check the SQL query statement to confirm the WHERE clause includes an expires_at condition.
If missing, add the condition: AND (expires_at IS NULL OR expires_at > ?).
Pass current as the third parameter in the execute call.
Run the test to verify.

Progressive hints

Hint 1: The current time is passed via the now parameter; if None, use datetime.now(timezone.utc).
Hint 2: In SQL, expires_at is stored as an ISO format string, so direct string comparison works.
Hint 3: Ensure the parameter order matches the placeholders in the SQL.

Success criteria

test_expired_memory_is_hidden_and_can_be_purged passes.
The list returned by query does not contain expired memories.

Fault injection and diagnosis

An expired session memory remains queryable and contaminates a later run unless expiry is enforced at read time. The executable fault fixture modifies forge/memory.py and is reliably reproduced by test_expired_memory_is_hidden_and_can_be_purged; the verified root cause is: The SELECT statement filters scope and subject but no longer requires expires_at to be null or later than now.

This stage’s core is memory boundaries, including expiry filtering. The fault is precisely due to the query method missing the expiry condition. The learner has already restored expiry filtering through guided practice, so they can identify and fix this defect.

Verified fault contract

Target test: test_expired_memory_is_hidden_and_can_be_purged
Faulty file: forge/memory.py

Diagnosis order

Run the failing test: python -m pytest -q ../tests/test_stage.py::test_expired_memory_is_hidden_and_can_be_purged
Observe the assertion failure: the returned list contains ‘expired’ and ‘valid’, but expected only ‘valid’.
Inspect the query method in forge/memory.py and find that the SQL lacks the expires_at condition.
Confirm the fix: add AND (expires_at IS NULL OR expires_at > ?) and bind current.

Fix principle: Add the expiry filtering condition AND (expires_at IS NULL OR expires_at > ?) to the SQL query in the query method, and pass current as the third parameter in the execute call. This way, the query excludes all expired memories and returns only currently valid records.

python -m pytest -q ../tests/test_stage.py::test_expired_memory_is_hidden_and_can_be_purged

Transfer challenge

In a music creation scenario, store a user preference separately from a single track-production record. The preference uses USER scope, while the production record uses RUN scope.

Changed condition: The data changes from software delivery to music creation while keeping the same MemoryStore API, scopes, and expiry rules.

Create a USER memory with subject_id ‘user-1’, content ‘prefers minor keys’, created_at 2026-08-17T09:00:00Z, and expires_at 2026-08-17T09:05:00Z. Create a non-expiring RUN memory with subject_id ‘run-1’ and content ‘draft saved’. Query USER/user-1 and RUN/run-1 with now=2026-08-17T10:00:00Z; the expired preference must be hidden, while the run memory must remain retrievable.

Success criteria

Querying USER scope for ‘user-1’ with now=2026-08-17T10:00:00Z returns an empty list because the memory expired at 2026-08-17T09:05:00Z.
Querying RUN scope for ‘run-1’ with the same now=2026-08-17T10:00:00Z returns the non-expiring memory whose content is ‘draft saved’.

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Design explanation question: Why do we need to distinguish RUN, SESSION, and USER scopes?

A strong answer should include

Different scopes correspond to different lifecycles and visibility ranges: RUN is valid only within a single run, SESSION within a session, and USER across sessions long-term.
Distinguishing scopes prevents information leakage and contamination: for example, a temporary decision from one run should not affect other runs or long-term user preferences.
Scopes also affect expiry policies: RUN memories may be automatically cleaned at run end, while USER memories may require explicit deletion or long-term retention.

Mastery evidence

All three tests pass.
The learner can explain memory boundaries, including scope, expiry, and deletion semantics.
The learner can correctly use MemoryStore to store and query memories of different scopes in the transfer challenge.

STAGE 06

Compress ordinary conversation while preserving instructions, permission decisions, and commitments verbatim with fidelity checks

Start with a concrete problem

How can we compress context without losing critical information?

In long tasks, the context window is limited, so ordinary messages must be compressed to save space. However, compression must not affect critical information: instructions, permission decisions, and commitments must be preserved verbatim, otherwise the system may perform unauthorized actions.

Currently, forge/context.py has not been created, and tests fail on import. You need to implement ContextPacket, CompressedContext, compress_context, and verify_critical_fidelity, ensuring that the compressed critical fields are exactly identical to the source, and any modification triggers a fidelity check failure.

Make a prediction

When implementing verify_critical_fidelity, what should happen if the compressed permission_decisions differ from the source?

Silently accept, because compression may change permissions
Raise ValueError, because permission decisions must be preserved verbatim
Log a warning but continue
Automatically restore the source permissions

Reasoning guide: The correct option is to raise ValueError. Permission decisions are critical context; any change indicates a security risk and must fail immediately. Silently accepting or auto-restoring could mask tampering.

What it is

Context compression fidelity model

Context compression summarizes ordinary messages into a short text while copying critical fields (instructions, permission decisions, commitments) verbatim into the compressed result.

Critical fields are bound by a SHA-256 fingerprint. Verification compares each field and checks the fingerprint to ensure no critical information was lost or tampered with during compression.

What it is not

Not indiscriminate summarization

Compression is not summarization of all content. Critical fields cannot be summarized because summarization may change semantics, e.g., turning “write_access=revoked” into “write access discussed”.

Compression is also not simple truncation. Ordinary messages may be truncated, but critical fields must remain complete, otherwise subsequent steps cannot rely on this information.

How it relates to neighboring concepts

Component relationships

ContextPacket is the source data containing all fields. compress_context generates CompressedContext, where critical fields are copied directly and ordinary messages are summarized.

verify_critical_fidelity receives the source and compressed result, compares critical fields and verifies the fingerprint. If any critical field mismatches or the fingerprint is inconsistent, it raises ValueError.

Boundary decision for this stage: Critical fields (instructions, permission decisions, commitments) must be preserved verbatim and fingerprinted; ordinary messages may be compressed. During verification, any difference in critical fields or fingerprint mismatch must cause failure.

Experiment: state-memory-context-06-context-compression
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/state-memory-context/06-context-compression/
Gate: Pending acceptance

Worked example

Goal: Demonstrate compressing context and verifying critical fields are preserved verbatim, ensuring compressed instructions, permission decisions, and commitments are identical to the source.

We start with a ContextPacket containing two instructions, one permission decision, one commitment, and 20 ordinary messages.

After calling compress_context, critical fields are copied verbatim, and ordinary messages are summarized into a string no longer than max_summary_chars.

Then verify_critical_fidelity is called; it compares critical fields and checks the fingerprint to ensure the compressed result is trustworthy.

Implementation steps

Create a ContextPacket instance with instructions, permission decisions, commitments, and messages.
Call compress_context with max_summary_chars=140.
Check that the compressed result’s instructions, permission_decisions, and commitments are the same as the source.
Check that message_summary length is <= 140 and source_message_count is 20.
Call verify_critical_fidelity and confirm no exception is raised.

Verified solution code: forge/context.py

from __future__ import annotations

import hashlib
import json

from pydantic import BaseModel, ConfigDict, Field


class ContextPacket(BaseModel):
    model_config = ConfigDict(frozen=True)

    instructions: tuple[str, ...] = Field(default_factory=tuple)
    permission_decisions: tuple[str, ...] = Field(default_factory=tuple)
    commitments: tuple[str, ...] = Field(default_factory=tuple)
    messages: tuple[str, ...] = Field(default_factory=tuple)


class CompressedContext(BaseModel):
    model_config = ConfigDict(frozen=True)

    instructions: tuple[str, ...]
    permission_decisions: tuple[str, ...]
    commitments: tuple[str, ...]
    message_summary: str
    source_message_count: int = Field(ge=0)
    critical_fingerprint: str = Field(min_length=64, max_length=64)


def _critical_payload(packet: ContextPacket) -> dict[str, tuple[str, ...]]:
    return {
        "instructions": packet.instructions,
        "permission_decisions": packet.permission_decisions,
        "commitments": packet.commitments,
    }


def critical_fingerprint(packet: ContextPacket) -> str:
    encoded = json.dumps(
        _critical_payload(packet),
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def _summarize_messages(messages: tuple[str, ...], max_chars: int) -> str:
    distinct: list[str] = []
    for message in messages:
        normalized = " ".join(message.split())
        if normalized and normalized not in distinct:
            distinct.append(normalized)
    if not distinct:
        return ""
    summary = " | ".join(distinct)
    if len(summary) <= max_chars:
        return summary
    marker = " ...[compressed]"
    return summary[: max_chars - len(marker)].rstrip() + marker


def compress_context(packet: ContextPacket, *, max_summary_chars: int = 400) -> CompressedContext:
    if max_summary_chars < 80:
        raise ValueError("max_summary_chars must leave enough room for a useful summary")
    return CompressedContext(
        instructions=packet.instructions,
        permission_decisions=packet.permission_decisions,
        commitments=packet.commitments,
        message_summary=_summarize_messages(packet.messages, max_summary_chars),
        source_message_count=len(packet.messages),
        critical_fingerprint=critical_fingerprint(packet),
    )


def verify_critical_fidelity(source: ContextPacket, compressed: CompressedContext) -> None:
    if compressed.instructions != source.instructions:
        raise ValueError("compression changed instructions")
    if compressed.permission_decisions != source.permission_decisions:
        raise ValueError("compression changed permission decisions")
    if compressed.commitments != source.commitments:
        raise ValueError("compression changed commitments")
    if compressed.critical_fingerprint != critical_fingerprint(source):
        raise ValueError("critical context fingerprint mismatch")

Verified solution code: forge/__init__.py

"""HeatStack Forge teaching package."""

from .context import CompressedContext, ContextPacket, compress_context, verify_critical_fidelity
from .memory import MemoryRecord, MemoryScope, MemoryStore

__all__ = [
    "CompressedContext",
    "ContextPacket",
    "MemoryRecord",
    "MemoryScope",
    "MemoryStore",
    "compress_context",
    "verify_critical_fidelity",
]

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Why this implementation

compress_context assigns critical fields directly to CompressedContext, so they are exactly the same as the source.

verify_critical_fidelity checks instructions, permission_decisions, commitments, and the fingerprint; all match, so it passes.

Guided practice

Restore permission decision comparison in verify_critical_fidelity to pass test_changed_permission_decision_fails_fidelity_check.

Look at the current implementation of verify_critical_fidelity in forge/context.py.
Notice it is missing the comparison for permission_decisions.
Add: if compressed.permission_decisions != source.permission_decisions: raise ValueError(“compression changed permission decisions”).
Run the test to confirm it passes.

Progressive hints

Compare compressed.permission_decisions and source.permission_decisions.
The error message should contain ‘permission decisions’ to match the test.

Success criteria

test_changed_permission_decision_fails_fidelity_check passes.
verify_critical_fidelity raises ValueError when permission decisions differ.

Fault injection and diagnosis

The failure fixture fixtures/failure.json describes a permission loss scenario: a summary drops the user decision that revoked write access. Critical context must remain verbatim and fingerprinted outside the lossy summary. In the current faulty code, verify_critical_fidelity skips the comparison of permission decisions, causing test_changed_permission_decision_fails_fidelity_check to fail. The executable faulty fixture modifies forge/context.py, and test_changed_permission_decision_fails_fidelity_check reliably reproduces the failure; the verified root cause is: verify_critical_fidelity checks other critical fields but no longer compares compressed.permission_decisions with source.permission_decisions. The executable fault fixture changes forge/context.py and is reproduced by test_changed_permission_decision_fails_fidelity_check; the verified root cause is: verify_critical_fidelity checks other critical fields but no longer compares compressed.permission_decisions with source.permission_decisions. The verified mechanism identifiers are write_access, verify_critical_fidelity, compressed.permission_decisions, source.permission_decisions, permission_decisions.

This stage has already implemented the basic structure of compress_context and verify_critical_fidelity, and the learner understands that critical fields must be preserved verbatim. The defect in the failure fixture is that verify_critical_fidelity lacks the check for permission_decisions, which is exactly the mechanism to be fixed in this stage. By restoring that check, we ensure permission decisions are not silently tampered with.

Verified fault contract

Target test: test_changed_permission_decision_fails_fidelity_check
Faulty file: forge/context.py

Diagnosis order

Run the failing test: python -m pytest -q ../tests/test_stage.py::test_changed_permission_decision_fails_fidelity_check
Observe output: Failed: DID NOT RAISE ValueError, indicating verify_critical_fidelity did not raise an exception.
Inspect the implementation of verify_critical_fidelity in forge/context.py and notice the missing comparison for permission_decisions.
Confirm that other critical fields (instructions, commitments) and the fingerprint are checked, but permission decisions are omitted.

Fix principle: Add a verbatim comparison for permission_decisions in verify_critical_fidelity: if compressed.permission_decisions != source.permission_decisions: raise ValueError(“compression changed permission decisions”).

python -m pytest -q ../tests/test_stage.py::test_changed_permission_decision_fails_fidelity_check

Transfer challenge

Software delivery scenario: you are adding a feature to a frontend repository, and the context includes code review instructions and permission decisions. After compressing the context, someone tampers with the permission decision, changing write_access=revoked to write_access=allowed.

Changed condition: The scenario data changes from generic test data to specific software delivery instructions and commitments, but the tampering target remains the permission decision (permission_decisions), not the instructions.

Use compress_context from forge.context to compress a ContextPacket containing software delivery instructions and commitments, where instructions are [“Review the frontend change before writing.”, “Keep test evidence for every modification.”], permission_decisions are [“write_access=revoked”], and commitments are [“Do not publish partial results.”]. Then tamper with the compressed result’s permission_decisions to [“write_access=allowed”] and call verify_critical_fidelity to verify it raises ValueError.

Success criteria

The compressed result’s critical fields match the source, and the source uses the specified instructions, permission_decisions, and commitments.
After tampering permission_decisions to write_access=allowed, verify_critical_fidelity raises ValueError containing ‘permission decisions’.

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Design explanation question: Why must critical context be preserved verbatim instead of summarized?

A strong answer should include

Summarization may change semantics, e.g., turning “write_access=revoked” into “write access discussed”, causing the system to misinterpret permission state.
Critical context (instructions, permission decisions, commitments) directly affects system behavior; any deviation may lead to security or compliance issues.
Verbatim preservation combined with fingerprint verification ensures that compression does not lose or tamper with critical information.

Mastery evidence

All three tests pass: test_compression_preserves_critical_context_verbatim, test_changed_permission_decision_fails_fidelity_check, test_fingerprint_prevents_silent_critical_context_replacement.
Learner can explain why critical context must be preserved verbatim and how the fingerprint prevents silent replacement.
Learner can apply compression and fidelity checks in a software delivery scenario and identify permission tampering.

STAGE 07

Replay a run from an auditable event journal and propagate memory deletion to dependent derived-summary caches

Start with a concrete problem

After deleting memory, derived summary caches still expose deleted content

In the previous stage, we implemented context compression, but the system still lacks an auditable event journal and a safe memory deletion mechanism. Currently, the file forge/replay.py does not exist, so tests fail at import time (ModuleNotFoundError: No module named ‘forge.replay’).

This stage requires implementing two core mechanisms: first, a hash-chained event journal that ensures the log is auditable and tamper-evident; second, when deleting memory, explicitly propagate the deletion to all derived summary caches that depend on that memory, preventing caches from continuing to expose deleted sensitive information.

Specifically, the test test_deletion_propagates_to_derived_summary_cache creates a user memory ‘private-note’ and generates two derived summaries: summary-1 references private-note, summary-2 does not. After calling delete_memory_everywhere, it expects summary-1 to be invalidated (invalidated == 1) while summary-2 remains unchanged.

The current failure fixture (fixtures/failure.json) shows that delete_memory_everywhere returns immediately after deleting the memory from the primary store without calling DerivedSummaryCache.invalidate_for_memory, causing the test to fail.

Make a prediction

When implementing delete_memory_everywhere, what happens if you only delete the memory from the primary store and do not handle the derived summary cache?

The derived summary cache updates automatically; no extra action is needed.
The derived summary cache still retains the deleted memory’s content, leading to data leakage.
The system throws an exception because the cache is inconsistent with the primary store.
The deletion fails because there are references in the cache.

Reasoning guide: The correct answer is the second option. The derived summary cache is stored independently and does not automatically detect changes in the primary store. Without explicit invalidation, the cache continues to expose the deleted memory content, violating data deletion integrity.

What it is

Relationship between event journal and derived summary cache

The event journal (EventJournal) is an append-only, hash-chained audit record. Each record contains the hash of the previous record (previous_hash) and the hash of the current event (record_hash). Any tampering causes a hash mismatch and is detected.

The derived summary cache (DerivedSummaryCache) stores summaries generated from multiple memories and records the list of source memory IDs (source_memory_ids) for each summary. When deleting a memory, you must find and delete all summaries that reference that memory; otherwise, the cache retains stale content.

The delete_memory_everywhere function coordinates the deletion: first delete the memory from the primary store, then call summary_cache.invalidate_for_memory(memory_id) to invalidate all related summaries and return the invalidation count.

What it is not

Common misconceptions

The event journal is not a plain text file; it uses a hash chain to guarantee integrity, and any modification breaks the chain.

The derived summary cache does not automatically synchronize with the primary store; explicit invalidation is required.

Deleting memory is not a single-point operation; it must propagate to all dependent derived data.

How it relates to neighboring concepts

Component collaboration

EventJournal uses RunEvent and reduce_events to replay state, ensuring that the state recovered from the journal matches the original state.

DerivedSummaryCache establishes dependencies between summaries and memories via the source_memory_ids field; the invalidate_for_memory method uses this field to find and delete related summaries.

delete_memory_everywhere connects MemoryStore and DerivedSummaryCache to implement cross-store deletion propagation.

Boundary decision for this stage: When deleting memory, you must handle both the primary store and all derived caches; deleting only from the primary store while ignoring caches leads to data inconsistency and potential leakage.

Experiment: state-memory-context-07-replay-and-deletion
Input: Software delivery: add a feature to a frontend repository, modify code, and add tests.
Evidence: evidence/state-memory-context/07-replay-and-deletion/
Gate: Pending acceptance

Worked example

Goal: Demonstrate creating an event journal, appending events, and replaying to verify state consistency.

First, create an EventJournal instance with a journal file path. Then construct three RunEvent objects representing run creation, step start, and write permission revocation.

Append the events to the journal using journal.append(event). Each append computes the current record’s hash and links it to the previous record.

Finally, call journal.replay() to read all events and replay them via reduce_events to obtain the final state. Verify that the replayed state matches the result of reduce_events(events) directly.

Implementation steps

Create an EventJournal instance with a temporary file path.
Construct three RunEvent objects with sequence numbers 1, 2, and 3.
Append each event using journal.append(event).
Call journal.read_events() to read all events and verify they equal the original event list.
Call journal.replay() and verify the returned RunState matches reduce_events(events).

Verified solution code: forge/replay.py

from __future__ import annotations

import hashlib
import json
import sqlite3
from pathlib import Path
from typing import Any

from .memory import MemoryStore
from .state.models import RunEvent, RunState, reduce_events

GENESIS_HASH = "0" * 64


def _record_hash(previous_hash: str, event_payload: dict[str, Any]) -> str:
    canonical = json.dumps(
        {"previous_hash": previous_hash, "event": event_payload},
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(canonical).hexdigest()


class EventJournal:
    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)

    def read_events(self) -> list[RunEvent]:
        if not self.path.exists():
            return []
        previous_hash = GENESIS_HASH
        events: list[RunEvent] = []
        for line_number, line in enumerate(self.path.read_text(encoding="utf-8").splitlines(), start=1):
            record = json.loads(line)
            if record.get("previous_hash") != previous_hash:
                raise ValueError(f"journal hash chain broken at line {line_number}")
            expected = _record_hash(previous_hash, record["event"])
            if record.get("record_hash") != expected:
                raise ValueError(f"journal record hash mismatch at line {line_number}")
            event = RunEvent.model_validate(record["event"])
            events.append(event)
            previous_hash = expected
        return events

    def append(self, event: RunEvent) -> str:
        events = self.read_events()
        if events:
            last = events[-1]
            if event.run_id != last.run_id:
                raise ValueError("one journal file may contain only one run_id")
            if event.sequence != last.sequence + 1:
                raise ValueError("journal events must use contiguous sequences")
            lines = self.path.read_text(encoding="utf-8").splitlines()
            previous_hash = json.loads(lines[-1])["record_hash"]
        else:
            if event.sequence != 1:
                raise ValueError("the first journal event must use sequence 1")
            previous_hash = GENESIS_HASH
        payload = event.model_dump(mode="json")
        record_hash = _record_hash(previous_hash, payload)
        record = {"previous_hash": previous_hash, "event": payload, "record_hash": record_hash}
        with self.path.open("a", encoding="utf-8", newline="\n") as handle:
            handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
        return record_hash

    def replay(self) -> RunState:
        return reduce_events(self.read_events())


class DerivedSummaryCache:
    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.connection = sqlite3.connect(self.path)
        self.connection.execute(
            """
            CREATE TABLE IF NOT EXISTS summaries (
                summary_id TEXT PRIMARY KEY,
                source_memory_ids TEXT NOT NULL,
                content TEXT NOT NULL
            )
            """
        )
        self.connection.commit()

    def close(self) -> None:
        self.connection.close()

    def put(self, summary_id: str, source_memory_ids: list[str], content: str) -> None:
        unique_ids = sorted(set(source_memory_ids))
        self.connection.execute(
            "INSERT OR REPLACE INTO summaries(summary_id, source_memory_ids, content) VALUES (?, ?, ?)",
            (summary_id, json.dumps(unique_ids), content),
        )
        self.connection.commit()

    def get(self, summary_id: str) -> str | None:
        row = self.connection.execute(
            "SELECT content FROM summaries WHERE summary_id = ?", (summary_id,)
        ).fetchone()
        return str(row[0]) if row else None

    def invalidate_for_memory(self, memory_id: str) -> int:
        rows = self.connection.execute("SELECT summary_id, source_memory_ids FROM summaries").fetchall()
        affected = [summary_id for summary_id, raw_ids in rows if memory_id in json.loads(raw_ids)]
        self.connection.executemany(
            "DELETE FROM summaries WHERE summary_id = ?", ((summary_id,) for summary_id in affected)
        )
        self.connection.commit()
        return len(affected)


def delete_memory_everywhere(
    memory_store: MemoryStore,
    summary_cache: DerivedSummaryCache,
    memory_id: str,
) -> tuple[bool, int]:
    deleted = memory_store.delete(memory_id)
    invalidated = summary_cache.invalidate_for_memory(memory_id)
    return deleted, invalidated

Verified solution code: forge/__init__.py

"""HeatStack Forge teaching package."""

from .context import CompressedContext, ContextPacket, compress_context, verify_critical_fidelity
from .memory import MemoryRecord, MemoryScope, MemoryStore
from .replay import DerivedSummaryCache, EventJournal, delete_memory_everywhere

__all__ = [
    "CompressedContext",
    "ContextPacket",
    "DerivedSummaryCache",
    "EventJournal",
    "MemoryRecord",
    "MemoryScope",
    "MemoryStore",
    "compress_context",
    "delete_memory_everywhere",
    "verify_critical_fidelity",
]

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Why this implementation

EventJournal.append checks event sequence continuity and computes the hash chain.

read_events verifies the hash chain while reading, ensuring the journal has not been tampered with.

replay uses reduce_events to fold the event sequence into a final state, enabling deterministic replay.

Guided practice

Add derived summary cache invalidation in delete_memory_everywhere to pass test_deletion_propagates_to_derived_summary_cache.

Open forge/replay.py and locate the delete_memory_everywhere function.
In the function, call memory_store.delete(memory_id) to delete the memory from the primary store.
Call summary_cache.invalidate_for_memory(memory_id) to invalidate all summaries referencing that memory and capture the invalidation count.
Return the tuple (deleted, invalidated).
Run the test to verify.

Progressive hints

The invalidate_for_memory method is already implemented; it finds summaries whose source_memory_ids contain memory_id and deletes them.
Ensure the returned invalidated is the actual number of summaries invalidated.

Success criteria

test_deletion_propagates_to_derived_summary_cache passes.
delete_memory_everywhere returns the correct (deleted, invalidated) tuple.
summary-1 is deleted, summary-2 remains unchanged.

Fault injection and diagnosis

The failure fixture fixtures/failure.json simulates a common defect: delete_memory_everywhere deletes the memory from the primary store but does not synchronously handle the derived summary cache, causing the cache to retain the deleted memory’s content. The test test_deletion_propagates_to_derived_summary_cache fails because it asserts invalidated == 1 but gets 0. Based on the test output and code analysis, diagnose and fix the defect. The executable faulty fixture modifies forge/replay.py and is reliably reproduced by test_deletion_propagates_to_derived_summary_cache; the verified root cause is: delete_memory_everywhere returns after MemoryStore.delete and never calls DerivedSummaryCache.invalidate_for_memory.

This stage already implements the DerivedSummaryCache.invalidate_for_memory method, which can find and delete all summaries referencing a given memory based on source_memory_ids. The learner only needs to call this method inside delete_memory_everywhere to fix the fault.

Verified fault contract

Target test: test_deletion_propagates_to_derived_summary_cache
Faulty file: forge/replay.py

Diagnosis order

Run the failing test: python -m pytest -q ../tests/test_stage.py::test_deletion_propagates_to_derived_summary_cache
Observe the assertion failure: assert invalidated == 1, but actual is 0.
Inspect the implementation of delete_memory_everywhere and analyze whether the necessary propagation step is missing after deleting from the primary store.
Confirm which methods DerivedSummaryCache provides for handling dependencies.

Fix principle: In delete_memory_everywhere, after memory_store.delete(memory_id), add invalidated = summary_cache.invalidate_for_memory(memory_id) and return (deleted, invalidated).

python -m pytest -q ../tests/test_stage.py::test_deletion_propagates_to_derived_summary_cache

Transfer challenge

In an office research scenario, a user requests deletion of a sensitive memory (e.g., a personal preference) that was used to generate a decision brief summary. You need to ensure the deletion operation also invalidates all summary caches that depend on that memory.

Changed condition: The scenario data changes from software delivery to office research; memory content and summary topics change accordingly, but the underlying mechanism remains the same.

Using the existing MemoryStore and DerivedSummaryCache, create a user memory (e.g., ‘private-note’), generate two summaries: one referencing that memory and one not. Call delete_memory_everywhere to delete the memory, verify the referencing summary is invalidated and the non-referencing summary remains unchanged.

Success criteria

delete_memory_everywhere returns (True, 1).
The summary referencing the deleted memory no longer exists (get returns None).
The summary not referencing the deleted memory retains its content.

Run the real tests

python -m pytest -q tests/test_stage.py

Run the real tests

...                                                                      [100%]
3 passed in <duration>s

Design explanation question: How does the hash chain detect journal tampering? Why do derived summary caches need explicit invalidation?

A strong answer should include

The hash chain links each record’s hash to the previous record; any modification to a historical record causes subsequent hashes to mismatch, thus detecting tampering.
Derived summary caches are independent of the primary store and do not automatically detect deletions; explicit invalidation is required to maintain data consistency.

Mastery evidence

All three tests pass (test_auditable_journal_replays_the_same_state, test_journal_tampering_is_detected_before_replay, test_deletion_propagates_to_derived_summary_cache).
Can explain the tamper-evident principle of the hash chain.
Can explain the necessity of propagating memory deletion to derived summary caches.

Complete the chapter

Pause, resume, and replay long tasks while controlling context growth.

FORGE / LOCAL CHECK

Local lab self-check

Not started
2Reading
3Lab downloaded
4Test result read
5Local check passed

verification.json is parsed only in this browser and is never uploaded. This checks the lab contract; it is not server certification or third-party endorsement.

Choose verification.json

Primary layer

Depends on

Provides to

Governed by

Concept calibration

Memory

Context

Observability

State, Sessions, Memory & Context Compression

Start with a concrete problem

How to rebuild current state from an event sequence?

Make a prediction

What it is

State is a projection of current progress, events are historical facts

What it is not

State is not history, and events are not mutable objects

How it relates to neighboring concepts

Events drive state transitions

Start with a concrete problem

Why do we need to distinguish sessions from runs?

Make a prediction

What it is

A session is a container for runs

What it is not

A session is not a single run

How it relates to neighboring concepts

Relationship among session, runs, and state

Start with a concrete problem

Saved checkpoints may be tampered with or version-incompatible

Make a prediction

What it is

Checkpoint Contract

What it is not

What a Checkpoint Is Not

How it relates to neighboring concepts

Relationships with Other Concepts

Start with a concrete problem

How can we prove that state recovered from a checkpoint matches a full event replay exactly?

Make a prediction

What it is

What recovery equivalence verification is

What it is not

What recovery equivalence verification is not

How it relates to neighboring concepts

Relationship with other concepts

Start with a concrete problem

Separate long-term memory from runtime state and ensure memory is retrievable only within correct scope and validity period

Make a prediction

What it is

Memory boundary model

What it is not

What memory boundaries are not

How it relates to neighboring concepts

Relationship between memory and other concepts

Start with a concrete problem

How can we compress context without losing critical information?

Make a prediction

What it is

Context compression fidelity model

What it is not

Not indiscriminate summarization

How it relates to neighboring concepts

Component relationships

Start with a concrete problem

After deleting memory, derived summary caches still expose deleted content

Make a prediction

What it is

Relationship between event journal and derived summary cache

What it is not

Common misconceptions

How it relates to neighboring concepts

Component collaboration