diff --git a/shepherd/packages/contexts/pyproject.toml b/shepherd/packages/contexts/pyproject.toml
index c280fef..28d3c1c 100644
--- a/shepherd/packages/contexts/pyproject.toml
+++ b/shepherd/packages/contexts/pyproject.toml
@@ -38,6 +38,7 @@ mcp = "shepherd_contexts.mcp:MCPServerContext"
 database = "shepherd_contexts.database:DatabaseContext"
 kvstore = "shepherd_contexts.kvstore:KVStoreContext"
 appstore = "shepherd_contexts.appstore:AppStoreContext"
+memory = "shepherd_contexts.memory:MemoryContext"
 
 [project.entry-points."shepherd.effects"]
 workspace = "shepherd_contexts.workspace.effects"
@@ -47,6 +48,7 @@ kvstore = "shepherd_contexts.kvstore.effects"
 mcp = "shepherd_contexts.mcp.effects"
 appstore = "shepherd_contexts.appstore.effects"
 database = "shepherd_contexts.database.effects"
+memory = "shepherd_contexts.memory.effects"
 
 [dependency-groups]
 test = [
diff --git a/shepherd/packages/contexts/src/shepherd_contexts/memory/__init__.py b/shepherd/packages/contexts/src/shepherd_contexts/memory/__init__.py
new file mode 100644
index 0000000..13057b2
--- /dev/null
+++ b/shepherd/packages/contexts/src/shepherd_contexts/memory/__init__.py
@@ -0,0 +1,48 @@
+"""Advisory memory context for Shepherd.
+
+Surfaces cross-run memory into a task's system prompt as a *logged* effect
+(:class:`MemoryRecalled`), so what influenced a run is auditable in the trace.
+Memory is advisory-only — it never enters the effect-replay fold or justifies a
+release. Backends are pluggable via the :class:`MemoryBackend` protocol;
+:class:`InMemoryBackend` is the deterministic, dependency-free default.
+
+Concrete backends that talk to an external memory substrate live out-of-tree:
+implement :class:`MemoryBackend` and pass an instance to
+``MemoryContext.create(...)``.
+
+Quick Start
+-----------
+    from shepherd_contexts.memory import (
+        InMemoryBackend,
+        MemoryContext,
+        MemoryHint,
+    )
+
+    backend = InMemoryBackend([MemoryHint(title="auth", content="use setup-token")])
+    memory = MemoryContext.create(backend, query="claude auth", project="shepherd")
+
+The write path is out-of-band: at settlement (select/discard) or on TaskFailed,
+build observations from the trace via :func:`observations_from_effects` and
+persist them through ``backend.save(...)``.
+"""
+
+from __future__ import annotations
+
+from shepherd_contexts.memory.backend import (
+    InMemoryBackend,
+    MemoryBackend,
+)
+from shepherd_contexts.memory.context import MemoryContext
+from shepherd_contexts.memory.effects import MemoryRecalled
+from shepherd_contexts.memory.types import MemoryHint, MemoryObservation
+from shepherd_contexts.memory.write import observations_from_effects
+
+__all__ = [
+    "InMemoryBackend",
+    "MemoryBackend",
+    "MemoryContext",
+    "MemoryHint",
+    "MemoryObservation",
+    "MemoryRecalled",
+    "observations_from_effects",
+]
diff --git a/shepherd/packages/contexts/src/shepherd_contexts/memory/backend.py b/shepherd/packages/contexts/src/shepherd_contexts/memory/backend.py
new file mode 100644
index 0000000..600d8ab
--- /dev/null
+++ b/shepherd/packages/contexts/src/shepherd_contexts/memory/backend.py
@@ -0,0 +1,103 @@
+"""Pluggable memory backends for :class:`~shepherd_contexts.memory.context.MemoryContext`.
+
+The :class:`MemoryBackend` protocol is the seam: Shepherd knows how to *surface*
+and *log* recalled memory; a backend decides where it comes from.
+``InMemoryBackend`` is the deterministic default (no external dependencies,
+ideal for tests).
+
+This module ships only the generic, dependency-free SPI and a reference
+in-memory backend. Concrete backends that talk to an external memory substrate
+(e.g. a durable cross-session store) are provided out-of-tree — implement the
+:class:`MemoryBackend` protocol and pass an instance to
+``MemoryContext.create(...)``.
+"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from .types import MemoryHint, MemoryObservation
+
+
+@runtime_checkable
+class MemoryBackend(Protocol):
+    """Read/write SPI for an advisory memory substrate."""
+
+    name: str
+
+    def recall(
+        self,
+        query: str,
+        *,
+        project: str | None = None,
+        n: int = 5,
+    ) -> list[MemoryHint]:
+        """Return up to ``n`` advisory hints relevant to ``query``.
+
+        Must be total: never raise. A backend that cannot answer returns ``[]``.
+        """
+        ...
+
+    def save(self, observation: MemoryObservation) -> str | None:
+        """Persist a memory-worthy observation; return its id, or ``None``.
+
+        Must be total: never raise. Out-of-band only — callers must have already
+        settled the run (select/release/discard) or confirmed a TaskFailed.
+        """
+        ...
+
+
+class InMemoryBackend:
+    """Deterministic in-process backend. The default; ideal for tests.
+
+    ``recall`` does naive case-insensitive substring matching of the query
+    against hint title+content, returning the top-``n`` matches (stable order).
+    ``save`` appends to the in-process store and returns a synthetic id.
+    """
+
+    def __init__(self, hints: list[MemoryHint] | None = None) -> None:
+        self._hints: list[MemoryHint] = list(hints or [])
+        self._saved: list[MemoryObservation] = []
+        self._counter = 0
+
+    @property
+    def name(self) -> str:
+        return "memory"
+
+    def recall(
+        self,
+        query: str,
+        *,
+        project: str | None = None,
+        n: int = 5,
+    ) -> list[MemoryHint]:
+        del project  # the naive backend does not partition by project
+        if not query.strip():
+            # No query: surface the most recent hints (recency-ish, stable).
+            return list(self._hints[-n:])
+        needle = query.lower()
+        terms = needle.split()
+        scored: list[tuple[int, int, MemoryHint]] = []
+        for idx, hint in enumerate(self._hints):
+            hay = (hint.title + " " + hint.content).lower()
+            hits = sum(1 for term in terms if term in hay)
+            if hits:
+                scored.append((hits, -idx, hint))  # more hits first, then earlier
+        scored.sort(key=lambda t: (t[0], t[1]), reverse=True)
+        return [h for _, _, h in scored[:n]]
+
+    def save(self, observation: MemoryObservation) -> str | None:
+        self._counter += 1
+        obs_id = f"inmem-{self._counter}"
+        self._saved.append(observation)
+        return obs_id
+
+    # Test/diagnostics helpers -------------------------------------------------
+
+    @property
+    def saved(self) -> list[MemoryObservation]:
+        """Observations written via ``save`` (test inspection)."""
+        return list(self._saved)
+
+
+__all__ = ["InMemoryBackend", "MemoryBackend"]
diff --git a/shepherd/packages/contexts/src/shepherd_contexts/memory/context.py b/shepherd/packages/contexts/src/shepherd_contexts/memory/context.py
new file mode 100644
index 0000000..d0a3ce3
--- /dev/null
+++ b/shepherd/packages/contexts/src/shepherd_contexts/memory/context.py
@@ -0,0 +1,252 @@
+"""MemoryContext: advisory cross-run memory as a logged, auditable effect.
+
+A read-only execution context that recalls advisory hints from a
+:class:`~shepherd_contexts.memory.backend.MemoryBackend` and surfaces them into
+a run's system prompt. The recall is recorded as a
+:class:`~shepherd_contexts.memory.effects.MemoryRecalled` effect in the trace, so
+``shepherd run trace`` shows exactly what memory influenced a run.
+
+Design (per the integration council's consensus):
+
+- **Advisory only.** Memory never enters ``state(t) = fold(apply_effect, …)``.
+  A run's correctness depends solely on its effect trace. Hints shape the
+  prompt; they never justify a release or mutate execution state.
+- **Logged, not injected invisibly.** ``MemoryRecalled`` is emitted at extract
+  time with each hint's backend digest, so a recall is replay-auditable. This
+  keeps the trace honest about what shaped the agent — the "the digest lies"
+  failure mode is avoided because the recall is *in* the trace, not a hidden
+  prompt mutation.
+- **Eager recall.** The backend is consulted once at ``create()`` (side effects
+  allowed there), so ``configure()`` stays pure — it only reads the already-
+  recalled hints to build the prompt addition. The backend *name* is captured
+  as a serializable field so ``MemoryRecalled.backend`` survives serialization
+  (a PrivateAttr would be dropped, lying about provenance).
+- **Pluggable backend.** ``InMemoryBackend`` (default, deterministic) or
+  any backend implementing :class:`MemoryBackend`. Either degrades to empty.
+
+Known limitations
+-----------------
+- ``MemoryRecalled`` is emitted at extract time (after execution). A run that
+  fails *before* the extract phase (e.g. ``execute_sdk`` raises) will not emit
+  one, so the dedicated audit record of which hints shaped that run is absent —
+  though ``ContextConfigured`` still records that a memory binding was active.
+  Emitting the recall before execution would require a framework hook to publish
+  effects at configure/prepare time.
+- Hints reach the prompt via ``ProviderBinding.system_prompt_additions``, the
+  same channel every context uses. Any device/container execution mode that
+  does not propagate ``system_prompt_additions`` (a framework-wide concern, not
+  specific to memory) will omit the hints there; the in-process path is unaffected.
+
+Example:
+    from shepherd_contexts.memory import MemoryContext, InMemoryBackend, MemoryHint
+
+    backend = InMemoryBackend([MemoryHint(title="...", content="...")])
+    memory = MemoryContext.create(backend, query="how to auth claude", project="shepherd")
+
+    with Scope() as scope:
+        scope.bind("memory", memory)
+        # ... run a task; its prompt carries the hints, its trace logs MemoryRecalled
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import TYPE_CHECKING, Any, ClassVar, Self
+
+from pydantic import BaseModel, ConfigDict
+from shepherd_core.types import (
+    ExecutionResult,
+    ProviderBinding,
+    ProviderCapabilities,
+    ReversibilityLevel,
+)
+from shepherd_runtime.context import Bindable
+
+from shepherd_contexts.memory.effects import MemoryRecalled
+from shepherd_contexts.memory.types import MemoryHint
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+    from pathlib import Path
+
+    from shepherd_core.effects import Effect
+    from shepherd_runtime.context import Sandbox
+
+    from shepherd_contexts.memory.backend import MemoryBackend
+
+_ADVISORY_HEADER = (
+    "## Advisory memory (advisory only — verify before relying; never a release justification)"
+)
+
+
+class MemoryContext(BaseModel, Bindable):
+    """Read-only advisory memory context backed by a :class:`MemoryBackend`."""
+
+    __binding_name__: ClassVar[str] = "memory"
+
+    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
+
+    # Serializable recalled state (set eagerly by create()). backend_name is a
+    # field (not a PrivateAttr) so MemoryRecalled.backend survives serialization.
+    query: str = ""
+    project: str | None = None
+    max_hints: int = 5
+    hints: tuple[MemoryHint, ...] = ()
+    backend_name: str = "none"
+    frozen_context_id: str | None = None
+
+    @property
+    def context_id(self) -> str:
+        # NOTE: this is a QUERY-IDENTITY address (query|project|max_hints), not
+        # a content address — two contexts with the same query but different
+        # recalled hints collide. Replay-auditability lives in MemoryRecalled's
+        # hint_digests, not here. See KVStoreContext for a content-addressed id.
+        if self.frozen_context_id:
+            return self.frozen_context_id
+        return f"memory:{self._compute_hash()[:12]}"
+
+    @property
+    def reversibility(self) -> ReversibilityLevel:
+        # Pure advisory input — no world mutation, mechanically a no-op to undo.
+        return ReversibilityLevel.AUTO
+
+    def __str__(self) -> str:
+        """Invisible in the prompt body; hints travel via system_prompt_additions."""
+        return ""
+
+    def __repr__(self) -> str:
+        n = len(self.hints)
+        proj = f", project={self.project!r}" if self.project else ""
+        return f"MemoryContext(query={self.query!r}, hints={n}{proj})"
+
+    def _compute_hash(self) -> str:
+        base = f"{self.query}|{self.project or ''}|{self.max_hints}"
+        return hashlib.sha256(base.encode()).hexdigest()
+
+    @classmethod
+    def create(
+        cls,
+        backend: MemoryBackend | None,
+        *,
+        query: str = "",
+        project: str | None = None,
+        max_hints: int = 5,
+    ) -> MemoryContext:
+        """Eagerly recall and build a context.
+
+        The backend is consulted here (side effects allowed) so the returned
+        context's ``configure()`` can be pure. A ``None`` backend yields an
+        empty (no-op) context.
+        """
+        ctx = cls(query=query, project=project, max_hints=max_hints)
+        if backend is not None and query.strip():
+            recalled = backend.recall(query, project=project, n=max_hints)
+            object.__setattr__(ctx, "hints", tuple(recalled))
+            object.__setattr__(ctx, "backend_name", backend.name)
+        frozen_id = f"memory:{ctx._compute_hash()[:12]}"
+        object.__setattr__(ctx, "frozen_context_id", frozen_id)
+        return ctx
+
+    # === ExecutionContext protocol ===
+
+    def configure(
+        self,
+        capabilities: ProviderCapabilities | None = None,
+    ) -> ProviderBinding:
+        """Pure: surface recalled hints as an advisory system-prompt addition."""
+        _ = capabilities  # memory is provider-agnostic
+        return ProviderBinding(
+            context_id=self.context_id,
+            context_type="MemoryContext",
+            context_description=f"Advisory memory ({len(self.hints)} hint(s))",
+            visible=False,  # invisible in the prompt body; hints via additions
+            system_prompt_additions=(self._prompt_block(),) if self.hints else (),
+        )
+
+    def prepare(self) -> MemoryContext:
+        """No-op — recall happened eagerly in ``create()``."""
+        return self
+
+    def cleanup(self, error: Exception | None = None) -> None:
+        """No resources to release."""
+        _ = error
+
+    def extract_effects(
+        self,
+        sandbox: Sandbox | None,
+        result: ExecutionResult,
+    ) -> Sequence[Effect]:
+        """Emit one ``MemoryRecalled`` recording what was surfaced (pure)."""
+        _ = sandbox, result
+        return (
+            MemoryRecalled(
+                query=self.query,
+                backend=self.backend_name,
+                project=self.project,
+                hint_count=len(self.hints),
+                hint_titles=tuple(h.title for h in self.hints),
+                hint_digests=tuple(h.digest for h in self.hints),
+                context_id=self.context_id,
+            ),
+        )
+
+    def apply_effect(self, effect: Effect) -> Self:
+        """Memory is read-only advisory — it derives no state from effects."""
+        _ = effect
+        return self
+
+    # === State serialization (device-boundary crossing) ===
+    # Memory is host-side advisory; it does not cross device boundaries, so
+    # transfer_bundle returns None (mirrors KVStoreContext). Defining these
+    # explicitly avoids an AttributeError if a checkpoint/transfer path invokes
+    # them, and makes the limitation explicit.
+
+    def to_state(self) -> dict[str, Any]:
+        """Serialize to a JSON-compatible state object."""
+        return {
+            "query": self.query,
+            "project": self.project,
+            "max_hints": self.max_hints,
+            "hints": [h.model_dump(mode="json") for h in self.hints],
+            "backend_name": self.backend_name,
+            "frozen_context_id": self.frozen_context_id,
+        }
+
+    @classmethod
+    def from_state(
+        cls,
+        state: Any,
+        sandbox_path: Path | str | None = None,
+    ) -> MemoryContext:
+        """Reconstruct from state. Memory has no filesystem state, so the path is ignored."""
+        _ = sandbox_path
+        if not isinstance(state, dict):
+            raise TypeError(f"MemoryContext.from_state expected dict, got {type(state).__name__}")
+        hints = tuple(
+            MemoryHint(**h) if isinstance(h, dict) else MemoryHint.model_validate(h)
+            for h in state.get("hints", [])
+        )
+        return cls(
+            query=state.get("query", ""),
+            project=state.get("project"),
+            max_hints=state.get("max_hints", 5),
+            hints=hints,
+            backend_name=state.get("backend_name", "none"),
+            frozen_context_id=state.get("frozen_context_id"),
+        )
+
+    def transfer_bundle(self, scope: Any) -> None:
+        """Memory stays on the host; it does not cross device boundaries."""
+        return
+
+    # === Helpers ===
+
+    def _prompt_block(self) -> str:
+        lines = [_ADVISORY_HEADER]
+        for hint in self.hints:
+            digest = f" (memory:{hint.digest})" if hint.digest else ""
+            lines.append(f"- [{hint.type}] {hint.title}: {hint.content}{digest}")
+        return "\n".join(lines)
+
+
+__all__ = ["MemoryContext"]
diff --git a/shepherd/packages/contexts/src/shepherd_contexts/memory/effects.py b/shepherd/packages/contexts/src/shepherd_contexts/memory/effects.py
new file mode 100644
index 0000000..208cdd7
--- /dev/null
+++ b/shepherd/packages/contexts/src/shepherd_contexts/memory/effects.py
@@ -0,0 +1,51 @@
+"""Memory context effects.
+
+``MemoryRecalled`` is the audit record for advisory memory surfaced into a run.
+It is emitted by :class:`~shepherd_contexts.memory.context.MemoryContext` at
+extract time and persisted in the effect trace, so ``shepherd run trace`` shows
+exactly which recalled hints influenced a run — resolving the "the digest lies"
+objection (a recall is part of the trace, not a hidden prompt mutation).
+
+Provenance: each recalled hint's ``digest`` (the backend's content address) is
+recorded, so a hint is replay-auditable across consolidation epochs.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Literal
+
+from shepherd_core.effects import Effect
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+
+class MemoryRecalled(Effect):
+    """Advisory memory was recalled and surfaced into a run's system prompt.
+
+    Emitted once per task execution that binds a MemoryContext, regardless of
+    whether the agent acted on the hints. Carries the query, the backend that
+    answered, the project scope, and a compact provenance record of the hints
+    (title + digest per hint) so the recall is auditable and replayable without
+    re-running the backend.
+    """
+
+    effect_type: Literal["memory_recalled"] = "memory_recalled"
+
+    # What was asked and who answered.
+    query: str = ""
+    backend: str = ""
+    project: str | None = None
+
+    # Compact provenance: per hint, its title and backend digest (if any).
+    hint_count: int = 0
+    hint_titles: tuple[str, ...] = ()
+    hint_digests: tuple[str | None, ...] = ()
+
+
+def get_effect_types() -> Mapping[str, type[Effect]]:
+    """Return the explicit effect contributor surface for runtime decode."""
+    return {"memory_recalled": MemoryRecalled}
+
+
+__all__ = ["MemoryRecalled", "get_effect_types"]
diff --git a/shepherd/packages/contexts/src/shepherd_contexts/memory/types.py b/shepherd/packages/contexts/src/shepherd_contexts/memory/types.py
new file mode 100644
index 0000000..58a9bd0
--- /dev/null
+++ b/shepherd/packages/contexts/src/shepherd_contexts/memory/types.py
@@ -0,0 +1,65 @@
+"""Data types for the memory context.
+
+Observation/hint shapes for a :class:`MemoryBackend`, compatible with a typical
+content-addressable memory substrate, without the framework depending on any
+specific one.
+
+Design notes
+------------
+- Hints are *advisory*: they surface into a run's system prompt but never enter
+  the ``state(t) = fold(apply_effect, effects)`` derivation. A run's correctness
+  depends only on its effect trace; recalled memory is logged (see
+  :class:`~shepherd_contexts.memory.effects.MemoryRecalled`) for auditability,
+  never treated as authoritative.
+- ``digest`` is the backend's content/provenance address for a hint (e.g. an
+  observation id / record digest). Recording it in the trace makes a recalled
+  hint replay-auditable across consolidation epochs.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict
+
+MemoryType = Literal["decision", "pattern", "discovery", "bugfix", "learning", "manual"]
+"""Observation type taxonomy."""
+
+
+class MemoryHint(BaseModel):
+    """A single recalled advisory hint surfaced into a run."""
+
+    model_config = ConfigDict(frozen=True)
+
+    title: str
+    content: str
+    type: MemoryType = "learning"
+    # Backend content/provenance address (observation id / digest).
+    # Recorded in MemoryRecalled so a hint is replay-auditable.
+    digest: str | None = None
+    # Backend that produced this hint (e.g. "memory", an external store id).
+    source: str = "memory"
+    # Optional relevance score from the backend (0..1). Not load-bearing.
+    score: float | None = None
+
+
+class MemoryObservation(BaseModel):
+    """A memory-worthy observation written at settlement / failure time.
+
+    Shape mirrors a typical ``/observations`` payload so a backend can forward
+    it unchanged. Written out-of-band — only after the trace is closed and the
+    review gate has settled (select/release/discard) or a TaskFailed fired.
+    """
+
+    model_config = ConfigDict(frozen=True)
+
+    type: MemoryType = "learning"
+    title: str
+    content: str
+    project: str | None = None
+    topic_key: str | None = None
+    # Free-form provenance: the shepherd run/trace this observation came from.
+    source: str | None = None
+
+
+__all__ = ["MemoryHint", "MemoryObservation", "MemoryType"]
diff --git a/shepherd/packages/contexts/src/shepherd_contexts/memory/write.py b/shepherd/packages/contexts/src/shepherd_contexts/memory/write.py
new file mode 100644
index 0000000..d7125e4
--- /dev/null
+++ b/shepherd/packages/contexts/src/shepherd_contexts/memory/write.py
@@ -0,0 +1,145 @@
+"""Write path: derive memory-worthy observations from a run's effect trace.
+
+Pure and side-effect-free — call this *out-of-band*, after a run has settled
+(select/release/discard) or after a TaskFailed. Feed the resulting
+:class:`MemoryObservation` objects to ``backend.save(...)``.
+
+The settlement *decision* (was a completed run selected or discarded?) is not in
+the effect stream — it is the human's action at the review gate. Pass it in via
+``disposition`` so the observation records the supervisor's judgment, which the
+council identified as the highest-signal memory input.
+
+Failures (TaskFailed) are always extracted: they are the canonical bugfix/root-
+cause memory and need no human decision to be worth remembering.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Literal
+
+from shepherd_contexts.memory.types import MemoryObservation
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+    from shepherd_core.effects import Effect
+
+Disposition = Literal["select", "discard", "release"]
+
+
+def observations_from_effects(
+    effects: Iterable[Effect],
+    *,
+    project: str | None = None,
+    disposition: Disposition | None = None,
+    source: str | None = None,
+) -> list[MemoryObservation]:
+    """Extract memory-worthy observations from a run's effects.
+
+    Args:
+        effects: The run's effect sequence (e.g. a scope's stream).
+        project: Project namespace for the observations.
+        disposition: The human's settlement decision for a *completed* run
+            (``select``/``release``/``discard``). Combined with TaskCompleted to
+            emit a decision/anti-pattern observation. Failures ignore this.
+        source: Provenance — the run/trace id these observations came from.
+
+    Returns:
+        Observations to persist via ``backend.save(...)``. Empty if nothing
+        memory-worthy was found.
+    """
+    out: list[MemoryObservation] = []
+    completed_tasks: list[str] = []
+
+    for effect in effects:
+        etype = getattr(effect, "effect_type", "")
+        if etype == "task_failed":
+            out.append(_observation_from_failure(effect, project=project, source=source))
+        elif etype == "task_completed":
+            # disposition applies per completed task (parallel to failures, so no
+            # completion is silently dropped). A nameless completion falls back to
+            # 'task' so it still yields an observation. Downstream topic_key dedupes.
+            completed_tasks.append(getattr(effect, "task_name", None) or "task")
+
+    # A completed run that the supervisor discarded is a high-signal anti-pattern;
+    # one that was selected is a validated decision. One observation per completed
+    # task (the human verdict applies to the run; topic_key dedupes repeats).
+    if disposition:
+        for task in completed_tasks:
+            out.append(
+                _observation_from_disposition(
+                    task,
+                    disposition=disposition,
+                    project=project,
+                    source=source,
+                )
+            )
+
+    return out
+
+
+def _observation_from_failure(
+    effect: Effect,
+    *,
+    project: str | None,
+    source: str | None,
+) -> MemoryObservation:
+    error = getattr(effect, "error", "") or "(no error message)"
+    error_type = getattr(effect, "error_type", "") or "error"
+    phase = getattr(effect, "phase", "") or ""
+    last_tool = getattr(effect, "last_tool_name", None)
+    loc = getattr(effect, "error_location", None)
+    suggestions = getattr(effect, "suggestions", ()) or ()
+    task = getattr(effect, "task_name", None) or "task"
+
+    lines = [error]
+    if phase:
+        lines.append(f"Failed in phase: {phase}")
+    if last_tool:
+        lines.append(f"Last tool: {last_tool}")
+    if loc:
+        lines.append(f"At: {loc}")
+    if suggestions:
+        lines.append("Suggestions: " + "; ".join(suggestions))
+
+    return MemoryObservation(
+        type="bugfix",
+        title=f"{task} failed: {error_type}",
+        content="\n".join(lines),
+        project=project,
+        topic_key=f"failure:{error_type}",
+        source=source or task,
+    )
+
+
+def _observation_from_disposition(
+    task: str,
+    *,
+    disposition: Disposition,
+    project: str | None,
+    source: str | None,
+) -> MemoryObservation:
+    if disposition == "discard":
+        return MemoryObservation(
+            type="pattern",
+            title=f"{task}: supervisor discarded the output",
+            content=(
+                f"A completed run of {task} was discarded at the review gate. "
+                "Treat the approach as suspect for similar future tasks."
+            ),
+            project=project,
+            topic_key=f"discarded:{task}",
+            source=source or task,
+        )
+    # select / release -> a validated decision worth recalling positively.
+    return MemoryObservation(
+        type="decision",
+        title=f"{task}: supervisor {disposition}ed the output",
+        content=f"A completed run of {task} was {disposition}ed at the review gate.",
+        project=project,
+        topic_key=f"{disposition}:{task}",
+        source=source or task,
+    )
+
+
+__all__ = ["Disposition", "observations_from_effects"]
diff --git a/shepherd/packages/contexts/tests/integration/test_memory_end_to_end.py b/shepherd/packages/contexts/tests/integration/test_memory_end_to_end.py
new file mode 100644
index 0000000..11a0af9
--- /dev/null
+++ b/shepherd/packages/contexts/tests/integration/test_memory_end_to_end.py
@@ -0,0 +1,90 @@
+"""End-to-end demo: MemoryContext recall -> task run -> logged MemoryRecalled effect.
+
+Exercises the full council wedge with the deterministic MockProvider (no LLM,
+no network): a task binds a MemoryContext, the recalled hints reach the
+provider binding's system prompt, and the run's effect trace contains a
+MemoryRecalled recording exactly what was surfaced. This is the "read path"
+proving the advisory layer is both *visible to the agent* and *auditable in
+the trace*.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel
+from shepherd_contexts.memory import (
+    InMemoryBackend,
+    MemoryContext,
+    MemoryHint,
+    MemoryRecalled,
+)
+from shepherd_runtime.scope import Scope
+from shepherd_runtime.task.authoring import Context, Output, task
+from shepherd_tests import MockProvider
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+    from shepherd_core.effects import Effect
+
+
+@task
+class AnswerFromMemory(BaseModel):
+    """Answer a question, optionally informed by recalled advisory memory."""
+
+    memory: Context(MemoryContext)
+    answer: Output(str)
+
+
+async def test_memory_wedge_recall_runs_and_logs() -> None:
+    backend = InMemoryBackend(
+        [
+            MemoryHint(
+                title="claude auth",
+                content="use CLAUDE_CODE_OAUTH_TOKEN for jailed runs",
+                digest="d1",
+                type="decision",
+            )
+        ]
+    )
+    memory = MemoryContext.create(backend, query="claude auth", project="shepherd")
+    provider = MockProvider(name="demo", mock_responses=[{"text": "use the oauth token"}])
+
+    async with Scope(root=True) as scope:
+        scope.bind("memory", memory)
+        scope.register_provider("default", provider, default=True)
+        await AnswerFromMemory.arun(scope=scope)
+        effects: Sequence[Effect] = [layer.effect for layer in scope.effects]
+
+    # 1) The recall is logged as a MemoryRecalled effect in the trace.
+    recalled = [e for e in effects if e.effect_type == "memory_recalled"]
+    assert len(recalled) == 1
+    assert isinstance(recalled[0], MemoryRecalled)
+    assert recalled[0].backend == "memory"
+    assert recalled[0].project == "shepherd"
+    assert recalled[0].hint_titles == ("claude auth",)
+    assert recalled[0].hint_digests == ("d1",)
+
+    # 2) The recalled hint reached the provider binding the agent was run under
+    #    (MemoryContext.configure() surfaced it via system_prompt_additions).
+    assert provider.calls, "MockProvider should have been invoked"
+    additions = provider.calls[0]["binding"].system_prompt_additions
+    assert any("CLAUDE_CODE_OAUTH_TOKEN" in a for a in additions)
+
+    # 3) The task completed (baseline: memory didn't break the run).
+    assert [e for e in effects if e.effect_type == "task_completed"]
+    # 4) Advisory-only: nothing mutated memory state. Effects attributed to the
+    #    memory binding are only read-only / lifecycle records — no state
+    #    mutation (no key_set / file_patch / similar) ever touches memory.
+    _benign = {
+        "memory_recalled",
+        "context_prepared",
+        "context_captured",
+        "context_cleaned_up",
+        "context_configured",
+    }
+    memory_binding_effects = [
+        e for e in effects if getattr(e, "binding_name", None) == "memory"
+    ]
+    assert all(e.effect_type in _benign for e in memory_binding_effects)
diff --git a/shepherd/packages/contexts/tests/unit/test_memory_context.py b/shepherd/packages/contexts/tests/unit/test_memory_context.py
new file mode 100644
index 0000000..9c4959b
--- /dev/null
+++ b/shepherd/packages/contexts/tests/unit/test_memory_context.py
@@ -0,0 +1,305 @@
+"""Tests for the advisory memory context.
+
+Covers the council's consensus wedge:
+- recall surfaces hints into the system prompt (read path)
+- the recall is logged as a MemoryRecalled effect (auditability)
+- memory is advisory-only (no state derivation, invisible body, AUTO reversibility)
+- backends are pluggable (InMemoryBackend default)
+- the write path derives memory-worthy observations from TaskFailed / settlement
+- MemoryRecalled round-trips through the composed effect registry (trace decode)
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from shepherd_contexts.memory import (
+    InMemoryBackend,
+    MemoryBackend,
+    MemoryContext,
+    MemoryHint,
+    MemoryObservation,
+    MemoryRecalled,
+    observations_from_effects,
+)
+from shepherd_core.effects.effects import TaskCompleted, TaskFailed
+from shepherd_core.types import ReversibilityLevel
+from shepherd_runtime.effects.registry import decode_effect
+
+if TYPE_CHECKING:
+    from shepherd_core.effects import Effect
+
+
+# =============================================================================
+# Helpers
+# =============================================================================
+
+
+def _hint(title: str, content: str = "", *, digest: str | None = None, hint_type: str = "learning") -> MemoryHint:
+    return MemoryHint(title=title, content=content, digest=digest, type=hint_type)  # type: ignore[arg-type]
+
+
+# =============================================================================
+# Read path: recall surfaces hints
+# =============================================================================
+
+
+class TestRecallSurfacesHints:
+    def test_create_recalls_matching_hints(self):
+        backend = InMemoryBackend(
+            [_hint("auth", "use setup-token", digest="d1"), _hint("unrelated", "noise")]
+        )
+        ctx = MemoryContext.create(backend, query="claude auth", project="shepherd")
+
+        assert len(ctx.hints) == 1
+        assert ctx.hints[0].title == "auth"
+        assert ctx.hints[0].digest == "d1"
+
+    def test_configure_injects_advisory_addition(self):
+        backend = InMemoryBackend([_hint("auth", "use setup-token", digest="d1")])
+        ctx = MemoryContext.create(backend, query="auth")
+
+        binding = ctx.configure()
+
+        assert binding.visible is False  # invisible body; hints via additions
+        assert len(binding.system_prompt_additions) == 1
+        block = binding.system_prompt_additions[0]
+        assert "Advisory memory" in block
+        assert "use setup-token" in block
+        assert "memory:d1" in block  # provenance digest surfaced
+
+    def test_configure_pure_and_stable(self):
+        ctx = MemoryContext.create(InMemoryBackend([_hint("a")]), query="a")
+        b1 = ctx.configure()
+        b2 = ctx.configure()
+        assert b1.system_prompt_additions == b2.system_prompt_additions
+
+    def test_empty_query_yields_no_hints(self):
+        ctx = MemoryContext.create(InMemoryBackend([_hint("a")]), query="")
+        assert ctx.hints == ()
+        assert ctx.configure().system_prompt_additions == ()
+
+    def test_none_backend_is_noop(self):
+        ctx = MemoryContext.create(None, query="anything")
+        assert ctx.hints == ()
+        assert ctx.configure().system_prompt_additions == ()
+        effs = ctx.extract_effects(None, None)
+        assert effs[0].backend == "none"
+
+
+# =============================================================================
+# Read path: the recall is logged as a MemoryRecalled effect
+# =============================================================================
+
+
+class TestRecallIsLogged:
+    def test_extract_emits_memory_recalled(self):
+        backend = InMemoryBackend(
+            [_hint("auth", "use setup-token", digest="d1"), _hint("auth2", "x", digest="d2")]
+        )
+        ctx = MemoryContext.create(backend, query="auth", project="shepherd")
+
+        effs = ctx.extract_effects(None, None)
+
+        assert len(effs) == 1
+        eff = effs[0]
+        assert isinstance(eff, MemoryRecalled)
+        assert eff.effect_type == "memory_recalled"
+        assert eff.query == "auth"
+        assert eff.project == "shepherd"
+        assert eff.backend == "memory"
+        assert eff.hint_count == 2
+        assert eff.hint_titles == ("auth", "auth2")
+        assert eff.hint_digests == ("d1", "d2")
+
+    def test_decode_round_trips_through_registry(self):
+        backend = InMemoryBackend([_hint("auth", "use setup-token", digest="d1")])
+        ctx = MemoryContext.create(backend, query="auth", project="shepherd")
+        original = ctx.extract_effects(None, None)[0]
+
+        data = original.model_dump(mode="json")
+        # The composed registry includes discovered contributors (memory_recalled).
+        restored = decode_effect(data)
+
+        assert isinstance(restored, MemoryRecalled)
+        assert restored.query == original.query
+        assert restored.hint_titles == original.hint_titles
+        assert restored.hint_digests == original.hint_digests
+
+
+# =============================================================================
+# Advisory-only guarantees (the reversibility invariant)
+# =============================================================================
+
+
+class TestAdvisoryOnly:
+    def test_apply_effect_is_noop(self):
+        ctx = MemoryContext.create(InMemoryBackend([_hint("a")]), query="a")
+        eff = ctx.extract_effects(None, None)[0]
+        # Memory derives no state from effects — applying any effect returns self.
+        assert ctx.apply_effect(eff) is ctx
+
+    def test_invisible_in_prompt_body(self):
+        ctx = MemoryContext.create(InMemoryBackend([_hint("a")]), query="a")
+        assert str(ctx) == ""
+
+    def test_reversibility_is_auto(self):
+        ctx = MemoryContext.create(InMemoryBackend(), query="x")
+        assert ctx.reversibility == ReversibilityLevel.AUTO
+
+    def test_context_id_is_stable(self):
+        a = MemoryContext.create(InMemoryBackend(), query="auth", project="p")
+        b = MemoryContext.create(InMemoryBackend(), query="auth", project="p")
+        assert a.context_id == b.context_id
+        c = MemoryContext.create(InMemoryBackend(), query="other", project="p")
+        assert c.context_id != a.context_id
+
+
+# =============================================================================
+# Backend pluggability
+# =============================================================================
+
+
+class TestBackends:
+    def test_inmemory_recall_scores_by_term_hits(self):
+        backend = InMemoryBackend(
+            [
+                _hint("auth token", "setup-token works"),
+                _hint("auth retry", "refresh needed"),
+                _hint("deploy", "unrelated"),
+            ]
+        )
+        hits = backend.recall("auth", n=5)
+        assert {h.title for h in hits} == {"auth token", "auth retry"}
+
+    def test_inmemory_save_returns_id_and_records(self):
+        backend = InMemoryBackend()
+        oid = backend.save(MemoryObservation(title="t", content="c"))
+        assert oid is not None
+        assert len(backend.saved) == 1
+        assert backend.saved[0].title == "t"
+
+    def test_protocol_conformance(self):
+        assert isinstance(InMemoryBackend(), MemoryBackend)
+
+
+# =============================================================================
+# Write path: observations from effects
+# =============================================================================
+
+
+class TestObservationsFromEffects:
+    def test_task_failed_yields_bugfix_observation(self):
+        effects: list[Effect] = [
+            TaskFailed(
+                task_name="WriteProgram",
+                error="permission denied",
+                error_type="PermissionError",
+                phase="execute",
+                last_tool_name="file_write",
+                error_location="provider.py:42",
+                suggestions=("check grant",),
+            )
+        ]
+        obs = observations_from_effects(effects, project="shepherd")
+
+        assert len(obs) == 1
+        assert obs[0].type == "bugfix"
+        assert "WriteProgram" in obs[0].title
+        assert "PermissionError" in obs[0].title
+        assert "permission denied" in obs[0].content
+        assert obs[0].topic_key == "failure:PermissionError"
+        assert obs[0].project == "shepherd"
+
+    def test_completed_with_discard_yields_anti_pattern(self):
+        effects: list[Effect] = [TaskCompleted(task_name="WriteProgram")]
+        obs = observations_from_effects(effects, disposition="discard")
+
+        assert len(obs) == 1
+        assert obs[0].type == "pattern"
+        assert obs[0].topic_key == "discarded:WriteProgram"
+        assert "discarded" in obs[0].content
+
+    def test_completed_with_select_yields_decision(self):
+        effects: list[Effect] = [TaskCompleted(task_name="WriteProgram")]
+        obs = observations_from_effects(effects, disposition="select")
+
+        assert len(obs) == 1
+        assert obs[0].type == "decision"
+        assert obs[0].topic_key == "select:WriteProgram"
+
+    def test_completed_without_disposition_yields_nothing(self):
+        # Without the human's settlement decision, a clean completion is not memory-worthy.
+        effects: list[Effect] = [TaskCompleted(task_name="WriteProgram")]
+        assert observations_from_effects(effects) == []
+
+    def test_empty_trace_yields_nothing(self):
+        assert observations_from_effects([]) == []
+
+
+# =============================================================================
+# Recall limits and empty-query recency
+# =============================================================================
+
+
+class TestRecallLimits:
+    def test_max_hints_truncates_recall(self):
+        backend = InMemoryBackend([_hint(f"term{i}") for i in range(10)])
+        ctx = MemoryContext.create(backend, query="term", max_hints=3)
+        assert len(ctx.hints) == 3
+
+    def test_empty_query_returns_most_recent(self):
+        backend = InMemoryBackend([_hint("a"), _hint("b"), _hint("c")])
+        # No query: surface the last n hints (recency-ish, stable order).
+        assert [h.title for h in backend.recall("", n=2)] == ["b", "c"]
+
+    def test_non_matching_query_yields_nothing(self):
+        backend = InMemoryBackend([_hint("alpha")])
+        ctx = MemoryContext.create(backend, query="zzz-no-match")
+        assert ctx.hints == ()
+        assert ctx.extract_effects(None, None)[0].hint_count == 0
+
+
+# =============================================================================
+# Mixed write-path effects (M3: multiple completions; M4: nameless completion)
+# =============================================================================
+
+
+class TestWritePathMixed:
+    def test_multiple_failures_each_yield_observation(self):
+        effects: list[Effect] = [
+            TaskFailed(task_name="A", error="e1", error_type="E1"),
+            TaskFailed(task_name="A", error="e2", error_type="E2"),
+        ]
+        obs = observations_from_effects(effects)
+        assert len(obs) == 2
+        assert all(o.type == "bugfix" for o in obs)
+
+    def test_multiple_completions_each_yield_disposition(self):
+        effects: list[Effect] = [
+            TaskCompleted(task_name="B"),
+            TaskCompleted(task_name="C"),
+        ]
+        obs = observations_from_effects(effects, disposition="discard")
+        assert len(obs) == 2  # M3: no completion silently dropped
+        assert all(o.type == "pattern" for o in obs)
+        assert {o.topic_key for o in obs} == {"discarded:B", "discarded:C"}
+
+    def test_nameless_completion_falls_back_to_placeholder(self):
+        # M4: a TaskCompleted with no task_name still yields an observation.
+        effects: list[Effect] = [TaskCompleted()]
+        obs = observations_from_effects(effects, disposition="select")
+        assert len(obs) == 1
+        assert obs[0].topic_key == "select:task"
+
+    def test_mixed_failures_completions_and_disposition(self):
+        effects: list[Effect] = [
+            TaskFailed(task_name="A", error="boom", error_type="Err"),
+            TaskCompleted(task_name="B"),
+            TaskCompleted(task_name="B"),  # duplicate name -> two obs (downstream dedupes)
+            TaskCompleted(task_name="C"),
+        ]
+        obs = observations_from_effects(effects, disposition="discard")
+        assert len(obs) == 4  # 1 failure + 3 discarded completions
+        assert sum(1 for o in obs if o.type == "bugfix") == 1
+        assert sum(1 for o in obs if o.type == "pattern") == 3