# Lab Spec — DD-14: Mastra Observability Primitives (Native Pattern + Read/Write Memory)

**Deep-Dive**: DD-14 Mastra (Observability Primitives)
**Duration**: 90 minutes
**Level**: Senior Engineer and above
**Language**: Python 3.10+ (TypeScript-style pseudocode — Python is used so the lab runs without a TS toolchain; the patterns transfer directly)
**Prerequisites**: Modules 0–12; DD-01–13; read `01-teaching-document.md` before starting.

> *Implement both observability patterns (wrapper and native) and both memory architectures (conflated and read/write split) in pure Python. Verify that the native pattern exposes internal decisions the wrapper cannot see, and that the read/write split makes write-gating a structural defense.*

---

## Objectives

After completing this lab you will be able to:

1. Implement the wrapper pattern (intercept calls, record spans) and the native pattern (components emit events as part of their contract) and demonstrate the observability gap the wrapper cannot close.
2. Implement the conflated memory store (one handle, policy-based write-gating) and the explicit read/write split (two interfaces, structural write-gating).
3. Verify the structural defense: an agent without the write-memory interface cannot write — not because a policy forbids it, but because there is no handle to call.
4. Defend why Mastra is the Module 10 reference (native emission) and the Module 4.3 reference (interface-level write-gating).
5. Draw the structural parallel to NemoClaw (DD-09): both remove the handle rather than writing a policy.

---

## Setup

```python
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Protocol, Any
import json
import time
```

No external dependencies. Pure standard library. The "model" is a stubbed function so you can test the patterns, not the model.

---

## Phase 1 — The Wrapper Pattern (observability applied externally)

Implement the wrapper pattern that most SDKs use — Module 1.4's `withObservability`.

```python
@dataclass
class Span:
    """A wrapper-pattern span: records a boundary crossing."""
    operation: str
    inputs: dict
    outputs: dict | None = None
    timestamp: float = field(default_factory=time.time)

class Tracer:
    """The wrapper's recorder. Only sees what crosses the boundary."""
    def __init__(self) -> None:
        self.spans: list[Span] = []
    def record(self, operation: str, inputs: dict, outputs: dict) -> None:
        self.spans.append(Span(operation=operation, inputs=inputs, outputs=outputs))

def with_observability(component: Any, tracer: Tracer) -> Any:
    """
    The wrapper pattern. Wraps a component so the tracer sees boundary crossings.
    The component itself knows nothing about tracing.
    """
    class Wrapped:
        def run(self, input_data: dict) -> dict:
            result = component.run(input_data)
            # The wrapper records WHAT CROSSED THE BOUNDARY — nothing more.
            tracer.record(operation="run", inputs=input_data, outputs=result)
            return result
    return Wrapped()
```

**Checkpoint**: build a plain component (below), wrap it with a tracer, run it, and verify the tracer recorded exactly one span with the inputs and outputs. The wrapper sees the boundary crossing — it does not see inside.

---

## Phase 2 — The Native Pattern (observability emitted as part of the contract)

Implement the native pattern — Mastra's contribution. The component owns its event emission.

```python
@dataclass
class Event:
    """
    A native-pattern event. Records an INTERNAL decision, not just a boundary
    crossing. This is the load-bearing difference from the wrapper's Span.
    """
    event_type: str        # "tool_selected" | "alternative_rejected" | "context_pruned" | "boundary_crossing"
    payload: dict          # the decision's details — WHY, not just WHAT
    timestamp: float = field(default_factory=time.time)

class NativeComponent:
    """
    Mastra-style native component. Owns its event emission.
    component.run(input) → { result, events[] }
    """
    def __init__(self, tools: dict[str, Any]):
        self.tools = tools

    def run(self, input_data: dict) -> dict[str, Any]:
        events: list[Event] = []

        # INTERNAL DECISION 1: which tool to pick (invisible to a wrapper)
        task = input_data.get("task", "")
        chosen_tool = None
        for tool_name in self.tools:
            if tool_name in task:
                chosen_tool = tool_name
                events.append(Event(
                    event_type="tool_selected",
                    payload={"tool": tool_name, "why": f"task references '{tool_name}'",
                             "alternatives_considered": list(self.tools.keys())}
                ))
                break

        # INTERNAL DECISION 2: reject alternatives (invisible to a wrapper)
        for tool_name in self.tools:
            if tool_name != chosen_tool:
                events.append(Event(
                    event_type="alternative_rejected",
                    payload={"tool": tool_name, "why": "task does not reference this tool"}
                ))

        # BOUNDARY CROSSING: execute the tool (visible to a wrapper)
        result = {"tool_called": chosen_tool, "output": "stubbed"}
        events.append(Event(
            event_type="boundary_crossing",
            payload={"operation": "tool_call", "tool": chosen_tool}
        ))

        # The component returns BOTH result AND events. Events are part of the contract.
        return {"result": result, "events": events}
```

**Checkpoint**: build a NativeComponent with three tools, run it, and verify the events list contains a `tool_selected` event, multiple `alternative_rejected` events, and a `boundary_crossing` event. Compare to the wrapper pattern: the wrapper would only record the `boundary_crossing`.

**The critical comparison**: run the same logical operation under both patterns. The wrapper's tracer records one span (the boundary crossing). The native component emits N events (the internal decisions PLUS the boundary crossing). The difference — `tool_selected` and `alternative_rejected` — is exactly the observability gap the wrapper cannot close.

---

## Phase 3 — The Conflated Memory Store (policy-based write-gating)

Implement the conflated memory store that most harnesses use — one handle for read and write.

```python
class ConflatedMemory:
    """
    The conflated store. One handle for read and write.
    Write-gating requires a POLICY layer bolted on top — advisory, not structural.
    """
    def __init__(self) -> None:
        self._store: list[dict] = []

    def read(self, query: str) -> list[dict]:
        return [entry for entry in self._store if query.lower() in str(entry).lower()]

    def write(self, entry: dict) -> bool:
        """Anyone with the handle can write. Policy would go HERE — advisory."""
        self._store.append(entry)
        return True

class PolicyGatedMemory(ConflatedMemory):
    """
    Policy-based write-gating. A check on the unified handle.
    ADVISORY: can be bypassed, misconfigured, or forgotten.
    """
    def __init__(self, allowed_writers: set[str]) -> None:
        super().__init__()
        self.allowed_writers = allowed_writers

    def write(self, entry: dict, writer_id: str) -> bool:
        if writer_id not in self.allowed_writers:
            return False  # policy denied
        return super().write(entry)
```

**Checkpoint**: build a ConflatedMemory, verify any agent with the handle can write. Then build a PolicyGatedMemory with an allowed-writers set, verify a disallowed writer is denied — but note the denial is a policy check on the same handle, not a structural absence.

---

## Phase 4 — The Explicit Read/Write Split (Mastra's structural defense)

Implement Mastra's explicit read/write split — separate interfaces, separately permissionable.

```python
class ReadMemoryInterface(Protocol):
    """The read interface. Agents that should not write get ONLY this."""
    def read(self, query: str) -> list[dict]: ...

class WriteMemoryInterface(Protocol):
    """The write interface. Withholding this is the structural defense."""
    def write(self, entry: dict) -> bool: ...

class MastraMemory:
    """
    Mastra-style explicit read/write split.
    Read and write are SEPARATE interfaces — separately permissionable.
    """
    def __init__(self) -> None:
        self._store: list[dict] = []

    def read(self, query: str) -> list[dict]:
        return [entry for entry in self._store if query.lower() in str(entry).lower()]

    def write(self, entry: dict) -> bool:
        self._store.append(entry)
        return True

class ReadOnlyAgent:
    """
    An agent that receives ONLY the read interface.
    It has NO handle to call write() — the structural defense.
    """
    def __init__(self, read_memory: ReadMemoryInterface):
        self.read_memory = read_memory
        # NOTE: no write_memory attribute. The handle does not exist.

    def act(self, query: str) -> list[dict]:
        return self.read_memory.read(query)

    def attempt_poison(self, entry: dict) -> str:
        """
        The agent TRIES to poison memory. But it has no write handle.
        This is the structural defense: there is nothing to call.
        """
        return "ERROR: agent has no write_memory interface — poisoning impossible"

class ReadWriteAgent:
    """
    An agent that receives BOTH interfaces. It CAN write.
    The decision to grant write is an interface-level decision, made at construction.
    """
    def __init__(self, read_memory: ReadMemoryInterface, write_memory: WriteMemoryInterface):
        self.read_memory = read_memory
        self.write_memory = write_memory

    def act(self, query: str) -> list[dict]:
        return self.read_memory.read(query)

    def write(self, entry: dict) -> bool:
        return self.write_memory.write(entry)
```

**Checkpoint**: build a MastraMemory. Construct a ReadOnlyAgent with only the read interface. Call `attempt_poison()` and verify it returns the "no write_memory interface" error — the agent literally has no `write` method to call. Then construct a ReadWriteAgent with both interfaces and verify it CAN write.

---

## Phase 5 — The NemoClaw Parallel (verify the structural principle)

Verify that Mastra's read/write split and NemoClaw's credential isolation are the same structural principle at different layers.

```python
class SandboxAgent:
    """
    NemoClaw-style sandboxed agent. Credentials live OUTSIDE the sandbox's reach.
    The sandboxed agent has NO handle to the credentials — same structural principle
    as Mastra's ReadOnlyAgent having no handle to writeMemory.
    """
    def __init__(self):
        # NOTE: no credentials attribute. The handle does not exist.
        self.tools = ["read_file", "write_file"]

    def act(self, task: str) -> str:
        return f"executed: {task}"

    def attempt_credential_leak(self) -> str:
        """
        The sandboxed agent TRIES to leak credentials. But it has no handle.
        Same structural defense as ReadOnlyAgent.attempt_poison().
        """
        return "ERROR: agent has no credentials attribute — leak impossible"

# THE PARALLEL VERIFICATION:
# ReadOnlyAgent has no write_memory → cannot poison.
# SandboxAgent has no credentials → cannot leak.
# Both are interface-level defenses: the handle does not exist.
```

**Checkpoint**: instantiate both a ReadOnlyAgent (from Phase 4) and a SandboxAgent. Call `attempt_poison()` and `attempt_credential_leak()` respectively. Verify both return the "impossible — no handle" error. The parallel is the point: two different layers (memory writes vs sandbox credentials), one structural principle (remove the handle, do not write a policy).

---

## Deliverables

Submit a single Python file `lab_solution.py` that:

- [ ] Defines Span, Tracer, with_observability (Phase 1 — wrapper pattern)
- [ ] Defines Event, NativeComponent (Phase 2 — native pattern)
- [ ] Defines ConflatedMemory, PolicyGatedMemory (Phase 3 — conflated store + policy gating)
- [ ] Defines MastraMemory, ReadOnlyAgent, ReadWriteAgent (Phase 4 — read/write split)
- [ ] Defines SandboxAgent (Phase 5 — NemoClaw parallel)
- [ ] Includes a `main()` function that:
  - [ ] Runs a plain component under the wrapper pattern and prints the single span recorded
  - [ ] Runs a NativeComponent and prints all events emitted (including internal decisions)
  - [ ] Verifies the wrapper recorded 1 event (boundary crossing) and the native component recorded N events (internal decisions + boundary crossing)
  - [ ] Builds a ConflatedMemory and verifies any agent can write
  - [ ] Builds a PolicyGatedMemory and verifies a disallowed writer is denied (policy-based)
  - [ ] Builds a MastraMemory, constructs a ReadOnlyAgent, and verifies attempt_poison() returns the "no write interface" error
  - [ ] Constructs a ReadWriteAgent and verifies it CAN write
  - [ ] Instantiates a SandboxAgent and verifies attempt_credential_leak() returns the "no credentials" error — the NemoClaw parallel

---

## Solution Key (key assertions)

```python
# Wrapper vs Native observability:
assert len(wrapper_tracer.spans) == 1                       # only boundary crossing
native_result = native_component.run({"task": "read auth.ts"})
assert len(native_result["events"]) > 1                     # internal decisions + boundary
event_types = [e.event_type for e in native_result["events"]]
assert "tool_selected" in event_types                        # invisible to wrapper
assert "alternative_rejected" in event_types                 # invisible to wrapper

# Conflated memory — anyone with handle can write:
assert conflated.write({"poison": True}) is True

# Policy gating — advisory, on the same handle:
assert policy_gated.write({"poison": True}, writer_id="untrusted") is False

# Mastra read/write split — structural:
result = readonly_agent.attempt_poison({"poison": True})
assert "no write_memory interface" in result                 # the handle does not exist
assert readwrite_agent.write({"data": "clean"}) is True      # has the handle

# NemoClaw parallel — same structural principle:
leak_result = sandbox_agent.attempt_credential_leak()
assert "no credentials" in leak_result                       # the handle does not exist
```

---

## Stretch Goals

1. **Native adapter for bring-your-own components.** Write a `NativeAdapter` class that wraps a non-native component and emits synthetic events for its internal decisions. Demonstrate that the adapter closes the observability gap — but note in a comment that the adapter's events are inferred, not intrinsic, so they can drift if the underlying component changes.

2. **Comparison harness.** Build a `ComparePatterns` class that runs the same task under both the wrapper and native patterns and prints a diff: which events the wrapper missed. Use this to demonstrate the observability gap quantitatively — "the wrapper missed N internal decisions."

3. **Three-level memory with interface-level gating.** Extend MastraMemory to three tiers (working, episodic, semantic — like OpenHarness DD-13), each with separate read and write interfaces. Construct an agent that can read all three tiers but write only to working memory. Verify the structural defense at each tier.

4. **The policy-bypass demonstration.** Show that the PolicyGatedMemory (Phase 3) can be bypassed by an agent that calls the parent class's write method directly (since policy gating is a subclass override). Contrast with Mastra's ReadOnlyAgent, which has no write method to call at all — the structural defense cannot be bypassed because there is nothing to bypass.

---

## What This Lab Proves

This lab is the concrete demonstration of why Mastra is the Module 10 reference and the Module 4.3 reference. By the end you have:

- A wrapper pattern that records only boundary crossings and a native pattern that records internal decisions — and a quantitative demonstration of the observability gap between them.
- A conflated memory store with advisory policy gating and a Mastra-style read/write split with structural gating — and a demonstration that the structural defense cannot be bypassed because there is no handle to call.
- A SandboxAgent that proves the NemoClaw parallel: two different layers (memory writes vs sandbox credentials), one structural principle (remove the handle).

The lab is the argument: observability belongs in the component, not around it. And the strongest write-gating defense is to withhold the interface, not to write a policy that says "do not misuse it." Both are interface-level decisions — the architectural move the course keeps returning to.