# Lab Spec — DD-17: PicoClaw — Turn-Boundary Trimming + Hardware Attack Surface **Deep-Dive**: DD-17 · **Duration**: 90 minutes · **Language**: Python 3.10+ · **Dependencies**: none (stdlib only) **Goal**: Build a minimal simulation of PicoClaw's turn-boundary context trimming and append-only JSONL memory. Then simulate the novel hardware attack surface: model an I2C bus with sensors, confirm that an injected agent can silently read sensors (the insufficiency of the write-only confirm gate), and implement the ZeroClaw-level autonomy fix. --- ## Setup Create `picoclaw_sim.py`. No external dependencies. Type hints required. Python 3.10+. ```python from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Optional import json, os, tempfile ``` --- ## Phase 1 — Append-only JSONL memory + turn-boundary trimming ### Step 1: Define the message and turn types ```python class Role(str, Enum): USER = "user" ASSISTANT = "assistant" TOOL = "tool" @dataclass class Message: role: Role content: str tool_call_id: Optional[str] = None # set when role=ASSISTANT with tool_call, or role=TOOL (the result) tool_name: Optional[str] = None @dataclass class Turn: """A complete turn: user msg -> LLM iterations (incl. tool calls + results) -> response.""" messages: list[Message] def token_count(self) -> int: # Simplified: 1 token per 4 chars return sum(len(m.content) // 4 for m in self.messages) ``` ### Step 2: The append-only JSONL store ```python class AppendOnlyJSONL: """PicoClaw's short-term memory: append-only, crash-safe, logical truncation via skip offset.""" def __init__(self, session_key: str, root: str = tempfile.mkdtemp()): self.jsonl_path = os.path.join(root, f"{session_key}.jsonl") self.meta_path = os.path.join(root, f"{session_key}.meta.json") # Initialize meta with skip=0 if not os.path.exists(self.meta_path): self._write_meta({"summary": "", "skip": 0}) def _write_meta(self, meta: dict) -> None: with open(self.meta_path, "w") as f: json.dump(meta, f) def _read_meta(self) -> dict: with open(self.meta_path) as f: return json.load(f) def append(self, msg: Message) -> None: """Append a message as one JSON line. NEVER modifies existing lines.""" with open(self.jsonl_path, "a") as f: f.write(json.dumps({"role": msg.role.value, "content": msg.content, "tool_call_id": msg.tool_call_id, "tool_name": msg.tool_name}) + "\n") def read_all(self) -> list[dict]: """Read all lines, skipping the first `skip` (logical truncation).""" skip = self._read_meta()["skip"] if not os.path.exists(self.jsonl_path): return [] with open(self.jsonl_path) as f: lines = f.readlines() # Ignore partial last line (crash-safety: a mid-write kill loses at most one line) parsed = [] for line in lines[skip:]: try: parsed.append(json.loads(line)) except json.JSONDecodeError: continue # partial line — ignore on recovery return parsed def truncate(self, n_skip: int) -> None: """Logical truncation: bump skip offset. Messages stay in the file but are skipped on read.""" meta = self._read_meta() meta["skip"] = max(meta["skip"], n_skip) self._write_meta(meta) ``` ### Step 3: The turn-boundary trimmer ```python def is_over_context_budget(turns: list[Turn], tool_def_tokens: int, output_reserve: int, budget: int) -> bool: msg_tokens = sum(t.token_count() for t in turns) return msg_tokens + tool_def_tokens + output_reserve > budget def trim_at_turn_boundaries(turns: list[Turn], tool_def_tokens: int, output_reserve: int, budget: int) -> list[Turn]: """Cut whole Turns from the front until under budget. NEVER splits a tool_call/result pair.""" while turns and is_over_context_budget(turns, tool_def_tokens, output_reserve, budget): turns.pop(0) # drop the oldest WHOLE turn return turns ``` ### Step 4: Verify the trimmer Write `verify_trimmer()` that: 1. Builds a session with 5 turns. Turn 2 contains a tool-call sequence: an `ASSISTANT` message with `tool_call_id="t1"` followed by a `TOOL` message with `tool_call_id="t1"` (the result). 2. Sets a budget that requires dropping 2 turns. 3. Calls `trim_at_turn_boundaries`. 4. Asserts: the trimmed list starts at Turn 3 (Turns 1 and 2 dropped as whole units). 5. Asserts: no `ASSISTANT` message with a `tool_call_id` is present without its matching `TOOL` result. (This is the load-bearing invariant — the tool_call/result pair is never split.) Run it: `python3 picoclaw_sim.py` should print `Phase 1 PASS`. --- ## Phase 2 — The novel hardware attack surface ### Step 1: Model the I2C bus ```python class Risk(str, Enum): LOW = "low" MEDIUM = "medium" HIGH = "high" @dataclass class Sensor: addr: int # 7-bit I2C address, 0x03-0x77 name: str value: str # e.g. "23.5C" — the physical-environment data class I2CBus: """Simulated I2C bus with sensors. Models PicoClaw's hardware surface.""" def __init__(self, sensors: list[Sensor]): # Validate 7-bit addressing for s in sensors: assert 0x03 <= s.addr <= 0x77, f"invalid 7-bit I2C address: {hex(s.addr)}" self.sensors = {s.addr: s for s in sensors} def read(self, addr: int) -> str: """READ path — PicoClaw as-built: UNGATED. Returns sensor value.""" if addr not in self.sensors: return f"" return self.sensors[addr].value def write(self, addr: int, data: bytes, confirm: bool) -> str: """WRITE path — PicoClaw as-built: confirm=True required.""" if not confirm: return "BLOCKED: hardware writes require confirm=True" if len(data) > 256: return "BLOCKED: I2C 256-byte cap exceeded" return f"wrote {len(data)} bytes to {hex(addr)}" ``` ### Step 2: Demonstrate the insufficiency Write `verify_attack_surface()` that: 1. Creates an I2C bus with 3 sensors: temperature (0x48), pressure (0x49), humidity (0x4A). 2. Simulates a PROMPT-INJECTED agent that reads all three sensors with NO confirm and NO approval. Confirm it succeeds and returns the physical-environment data. 3. Simulates the same agent attempting a write with `confirm=False`. Confirm it is blocked. 4. Print the asymmetry: reads succeed silently; writes are gated. This is the insufficiency. ### Step 3: The ZeroClaw-level autonomy fix Implement an autonomy-gated I2C bus that treats ALL hardware interactions as risk-classified: ```python class AutonomyGatedI2CBus: """The fix: ZeroClaw-level autonomy gating on ALL hardware interactions (DD-16).""" def __init__(self, bus: I2CBus, autonomy: str = "supervised"): self.bus = bus self.autonomy = autonomy # "readonly", "supervised", "full" self.audit_log: list[dict] = [] def _gate(self, operation: str, addr: int, risk: Risk) -> tuple[bool, str]: """Returns (allowed, reason). All hardware ops are risk-classified.""" self.audit_log.append({"op": operation, "addr": hex(addr), "risk": risk.value}) if self.autonomy == "readonly": return False, "readonly autonomy: hardware access blocked" if self.autonomy == "supervised": # In real PicoClaw+ZeroClaw: this would route to an approver channel. # Here we simulate: MEDIUM (read) and HIGH (write) both need approval. return False, f"supervised autonomy: {operation} is {risk.value} risk — needs approval" # full autonomy return True, "full autonomy: allowed" def read(self, addr: int, approved: bool = False) -> str: allowed, reason = self._gate("read", addr, Risk.MEDIUM) if not allowed and not approved: return f"BLOCKED: {reason}" return self.bus.read(addr) def write(self, addr: int, data: bytes, approved: bool = False) -> str: allowed, reason = self._gate("write", addr, Risk.HIGH) if not allowed and not approved: return f"BLOCKED: {reason}" return self.bus.write(addr, data, confirm=True) ``` Write `verify_fix()` that: 1. Creates an `AutonomyGatedI2CBus` with `autonomy="supervised"`. 2. Confirms a prompt-injected agent CANNOT silently read sensors — the read is blocked pending approval. 3. Confirms that with explicit approval, the read succeeds. 4. Confirms the audit log records every hardware interaction (op, addr, risk) — the missing audit trail in PicoClaw as-built. Run it: `python3 picoclaw_sim.py` should print `Phase 2 PASS`. --- ## Phase 3 — The crash-safety property Write `verify_crash_safety()` that: 1. Creates an `AppendOnlyJSONL` store. Appends 5 valid messages. 2. Simulates a mid-write crash: appends a partial (invalid JSON) line to the `.jsonl` file directly (simulating a process kill mid-write). 3. Calls `read_all()`. Confirms the 5 valid messages are returned and the partial line is silently ignored. 4. Confirms no existing line was corrupted. **Discussion prompt**: Write one paragraph explaining why append-only + logical-truncation (skip offset) is safer than in-place mutation + physical deletion. Save your answer to `answers.md`. --- ## Phase 4 (stretch) — The LoCoMo-style evaluation Write a minimal memory-benchmark harness that: 1. Generates a synthetic long conversation (100+ turns with references to earlier turns). 2. Stores it in an `AppendOnlyJSONL` store. 3. Trims to a budget that drops 50% of turns. 4. Evaluates: can a simulated agent answer questions about facts from dropped turns? (It should not be able to — those facts are gone from the active context.) 5. Reports: what fraction of facts survive trimming as a function of where they appeared in the conversation. This is a miniature version of what PicoClaw's `cmd/membench` does with the LoCoMo benchmark. --- ## Solution key ### Phase 1 expected output ``` Built session with 5 turns; Turn 2 has a tool_call/result pair. Budget requires dropping 2 turns. After trim: 3 turns remain (Turns 3, 4, 5). Invariant check: every ASSISTANT tool_call has its matching TOOL result: True Phase 1 PASS ``` ### Phase 2 expected output ``` === PicoClaw AS-BUILT (confirm on writes only) === Prompt-injected agent reads temperature (0x48): 23.5C — UNGATED Prompt-injected agent reads pressure (0x49): 1013hPa — UNGATED Prompt-injected agent reads humidity (0x4A): 45% — UNGATED Prompt-injected agent writes with confirm=False: BLOCKED — gated ASYMMETRY: reads succeed silently; writes are gated. INSUFFICIENT. === ZEROCLAW-LEVEL FIX (autonomy gating on all hw) === Prompt-injected agent reads temperature (0x48): BLOCKED: supervised autonomy: read is medium risk — needs approval Approved read of temperature (0x48): 23.5C Audit log entries: 2 Phase 2 PASS ``` ### Phase 3 expected behavior - 5 valid messages appended. - Partial line injected (simulating mid-write crash). - `read_all()` returns the 5 valid messages; the partial line is silently ignored. - No existing line corrupted. --- ## Stretch goals 1. **Add hooks.** Implement PicoClaw's three hook types (observer 500ms, interceptor 5s, approval 60s) as callbacks on tool calls. Let the interceptor block specific tool names. 2. **Add the MaixCAM channel.** Simulate a TCP-socket channel that receives image frames (as base64 strings) and dispatches them to a vision-model callback. 3. **Implement the skip-offset compaction.** When the skip offset exceeds 50% of the file, rewrite the .jsonl without the skipped messages and reset skip to 0. Confirm this is safe (the rewrite happens in a temp file, then atomically renames). 4. **Cross-compilation simulation.** Document (do not implement) the build flags for an edge deployment: which Cargo features / Go build tags would you disable to shed the thick ecosystem? List the kernel-only tool set. --- ## Deliverables - `picoclaw_sim.py` — the simulator (Phases 1, 2, 3). - `answers.md` — your Phase 3 discussion paragraph (append-only vs in-place mutation). - (Stretch) `membench.py` — the Phase 4 evaluation harness. --- ## References - DD-17 teaching document, Section "The Hardware Differentiator" — the I2C/SPI/Serial validation rules. - DD-17 teaching document, Section "The Novel Hardware Attack Surface" — the write/read asymmetry. - DD-16 (ZeroClaw) — the autonomy-level model that is the fix for PicoClaw's hardware tools. - Module 3 (Context Management) — the turn-boundary trimming principle. - Module 4 (Memory) — append-only JSONL as a crash-safe short-term store. - Module 5 (Sandboxing) — the bwrap/restricted-token model and its macOS gap.