Harness Compiler API

This page renders the harness and compiler source APIs directly from code. It is the closest analogue to tinygrad's source-backed developer pages.

Capture

capture

Capture coordination and the lineage report (design §28.1 control ops, §51).

This is the control-plane logic behind both the CLI and the MCP server: start / finish a capture, mark an episode outcome, summarize, and produce the deterministic exact-lineage report that §51 names as the first concrete milestone::

frontier host -> jdsl capture -> canonical trace -> exact lineage report

The coordinator never depends on MCP; the MCP server is a thin shell over it.

Classes:

• CaptureCoordinator –

  Owns capture lifecycle and analysis over a HarnessStore.

CaptureCoordinator dataclass

CaptureCoordinator(store: HarnessStore)

Owns capture lifecycle and analysis over a HarnessStore.

Methods:

• lineage_report –

  The §51 exact-lineage report over a capture's episodes.

lineage_report

lineage_report(capture_id: str) -> dict[str, Any]

The §51 exact-lineage report over a capture's episodes.

Source code in jdsl_harness/capture.py

def lineage_report(self, capture_id: str) -> dict[str, Any]:
    """The §51 exact-lineage report over a capture's episodes."""
    episodes = [e for e in self.store.capture_episodes(capture_id)
                if not e.episode_id.startswith("_")]
    norm = normalize_all(episodes)
    per_episode = [_episode_lineage(n) for n in norm]
    candidates = consolidate(norm)
    deterministic = [c.to_dict() for c in candidates
                     if c.type in ("DATAFLOW", "CONTROL") and c.status != "contested"]
    residual = [c.to_dict() for c in candidates if c.type == "SEMANTIC"]
    return {
        "capture_id": capture_id,
        "episodes": per_episode,
        "deterministic_candidates": deterministic,
        "residual_candidates": residual,
    }

store

Harness storage (design §30, §35 PR6).

Boring storage first: SQLite for metadata, JSONL for the append-only event stream, a content-addressed blob store on the filesystem. The source of truth for raw execution is the append-only events + blobs; the SQLite tables are a rebuildable index over them (§30). No vector DB, no server dependency.

Layout under root/::

harness.db                 SQLite metadata (captures, episodes, event_index, packages)
captures/<capture_id>.jsonl append-only canonical event spool
blobs/sha256/<digest>      content-addressed blobs

Classes:

• HarnessStore –

  Metadata + event spool + blob store for one harness instance.

HarnessStore

HarnessStore(root: str | Path)

Metadata + event spool + blob store for one harness instance.

Methods:

• ingest –

  Record a single already-formed event (used by adapters/ingest server).

• sink –

  A trace sink that appends to this capture's spool and indexes each event.

Source code in jdsl_harness/store.py

def __init__(self, root: str | Path) -> None:
    self.root = Path(root)
    self.root.mkdir(parents=True, exist_ok=True)
    (self.root / "captures").mkdir(exist_ok=True)
    self.blobs = BlobStore(self.root / "blobs")
    self._db_path = self.root / "harness.db"
    self._local = threading.local()
    self._chainers: dict[str, _Chainer] = {}
    self._lock = threading.Lock()
    self._init_db()

ingest

ingest(event: TraceEvent) -> TraceEvent

Record a single already-formed event (used by adapters/ingest server).

Source code in jdsl_harness/store.py

def ingest(self, event: TraceEvent) -> TraceEvent:
    """Record a single already-formed event (used by adapters/ingest server)."""
    return self.sink(event.capture_id).emit(event)

sink

sink(capture_id: str) -> JsonlTraceSink

A trace sink that appends to this capture's spool and indexes each event. Reuses one hash-chainer per capture so sequences/hashes stay consistent.

Source code in jdsl_harness/store.py

def sink(self, capture_id: str) -> JsonlTraceSink:
    """A trace sink that appends to this capture's spool and indexes each event.
    Reuses one hash-chainer per capture so sequences/hashes stay consistent."""
    with self._lock:
        chain = self._chainers.setdefault(capture_id, _Chainer())
    return _IndexingSink(self, capture_id, JsonlTraceSink(self.spool_path(capture_id), chain=chain))

gateway

The jdsl tool gateway — Tier A capture (design §8.1, §35 PR5).

The preferred capture mode: the model sees task tools through jdsl, so the gateway records tool identity, schema, arguments, result, error, timing, and state around mutations (§8.1). Wrapping a tool with the gateway records a canonical tool.call.started/completed/failed triple on the sink, with the environment's state snapshot attached around state-changing calls when an adapter is present.

Classes:

• ToolGateway –

  Wraps callables/Tools so their calls are recorded to a trace sink. Preserves

ToolGateway

ToolGateway(sink: TraceSink, *, capture_id: str, episode_id: str, env: EnvironmentAdapter | None = None, source: EventSource | None = None)

Wraps callables/Tools so their calls are recorded to a trace sink. Preserves the original tool's schema and return value; capture is transparent to the caller (fail-open, §7.2 — a broken sink never breaks the tool call).

Methods:

• record_outcome –

  Emit the environment's task outcome, if the adapter reports one (§8.1).

• wrap –

  Return a callable that records each invocation. logical_id is the

Source code in jdsl_harness/gateway.py

def __init__(self, sink: TraceSink, *, capture_id: str, episode_id: str,
             env: EnvironmentAdapter | None = None,
             source: EventSource | None = None) -> None:
    self.sink = sink
    self.capture_id = capture_id
    self.episode_id = episode_id
    self.env = env
    self.source = source or EventSource(adapter="gateway")
    self._seq = 0

record_outcome

record_outcome() -> TraceEvent | None

Emit the environment's task outcome, if the adapter reports one (§8.1).

Source code in jdsl_harness/gateway.py

def record_outcome(self) -> TraceEvent | None:
    """Emit the environment's task outcome, if the adapter reports one (§8.1)."""
    if self.env is None:
        return None
    outcome = self.env.outcome()
    if outcome is None:
        return None
    return self._emit(EventKind.ENVIRONMENT_VERDICT, actor="environment",
                      payload=outcome.to_payload())

wrap

wrap(fn: Any, *, logical_id: str | None = None, host_name: str | None = None, destructive: bool = False) -> Any

Return a callable that records each invocation. logical_id is the portable capability id; destructive triggers state snapshots (§8.1).

Source code in jdsl_harness/gateway.py

def wrap(self, fn: Any, *, logical_id: str | None = None, host_name: str | None = None,
         destructive: bool = False) -> Any:
    """Return a callable that records each invocation. `logical_id` is the
    portable capability id; `destructive` triggers state snapshots (§8.1)."""
    name = host_name or getattr(fn, "name", None) or getattr(fn, "__name__", "tool")
    logical = logical_id or (self.env.canonical_tool(name) if self.env else name)

    def recorded(*args: Any, **kwargs: Any) -> Any:
        arguments = dict(kwargs)
        for i, a in enumerate(args):
            arguments[f"_arg{i}"] = a
        before = self._maybe_snapshot(destructive)
        started = self._emit(EventKind.TOOL_CALL_STARTED, actor="model", payload={
            "tool": {"logical_id": logical, "host_name": name}, "arguments": arguments,
        }, state_before=before)
        t0 = time.monotonic()
        try:
            result = fn(*args, **kwargs)
        except Exception as err:  # noqa: BLE001 — record then re-raise
            self._emit(EventKind.TOOL_CALL_FAILED, actor="tool",
                       parent_event_id=started.event_id, payload={
                           "tool": {"logical_id": logical, "host_name": name},
                           "error": str(err), "duration_ms": _ms(t0)})
            raise
        after = self._maybe_snapshot(destructive)
        kind = EventKind.TOOL_CALL_FAILED if _is_error(result) else EventKind.TOOL_CALL_COMPLETED
        self._emit(kind, actor="tool", parent_event_id=started.event_id, payload={
            "tool": {"logical_id": logical, "host_name": name},
            ("error" if kind == EventKind.TOOL_CALL_FAILED else "result"): result,
            "duration_ms": _ms(t0)}, state_after=after)
        return result

    recorded.logical_id = logical  # type: ignore[attr-defined]
    recorded.name = name  # type: ignore[attr-defined]
    return recorded

server

Harness daemon: local ingest data plane + MCP control plane (design §7, §28, §35 PR6).

The architectural split of §7 lives here. The data plane is a fast local HTTP loopback endpoint that host hooks POST events to and returns immediately (§7.2) — never a slow remote request on every tool event. The control plane (§7.1, §28) is a small, stable set of operations (jdsl.capture.*, jdsl.compile, …) exposed as an MCP server when the MCP SDK is available, and always available in-process via CaptureCoordinator.

MCP is optional: importing this module never requires the mcp package. If it is absent, build_mcp_server raises a clear error and the HTTP ingest server still runs.

Classes:

• IngestServer –

  Loopback HTTP ingest for host hooks and adapters (§7.2 telemetry plane).

Functions:

• build_mcp_server –

  Build an MCP server exposing the control tools (§28.1). Requires the mcp

IngestServer

IngestServer(store: HarnessStore, *, host: str = '127.0.0.1', port: int = 8848)

Loopback HTTP ingest for host hooks and adapters (§7.2 telemetry plane).

Endpoints

POST /ingest body: a canonical TraceEvent dict POST /hook/claude?cap=… body: a Claude Code hook payload POST /hook/gemini?cap=… body: a Gemini CLI hook payload POST /hook/opencode?cap=… body: a jdsl OpenCode hook envelope GET /captures list captures GET /capture//summary

The hook itself fails open for observation (§7.2): a bad request never 500s the agent — it returns a 200 with an error note so the host loop keeps moving.

Source code in jdsl_harness/server.py

def __init__(self, store: HarnessStore, *, host: str = "127.0.0.1", port: int = 8848) -> None:
    self.store = store
    self.coord = CaptureCoordinator(store)
    self.host = host
    self.port = port
    self._httpd: ThreadingHTTPServer | None = None
    self._thread: threading.Thread | None = None
    self._claude_correlator = ToolCallCorrelator()
    self._gemini_correlator = ToolCallCorrelator()
    self._opencode_correlator = ToolCallCorrelator()

build_mcp_server

build_mcp_server(store: HarnessStore, name: str = 'jdsl-harness') -> Any

Build an MCP server exposing the control tools (§28.1). Requires the mcp package; raises a clear error if it is not installed. Kept import-optional so the core harness never depends on the MCP SDK (§36).

Source code in jdsl_harness/server.py

def build_mcp_server(store: HarnessStore, name: str = "jdsl-harness") -> Any:
    """Build an MCP server exposing the control tools (§28.1). Requires the `mcp`
    package; raises a clear error if it is not installed. Kept import-optional so
    the core harness never depends on the MCP SDK (§36)."""
    coord = CaptureCoordinator(store)
    mcp = _mcp_server(name)

    @mcp.tool()
    def jdsl_capture_start(host: str = "jdsl", adapter: str = "runtime", note: str = "") -> dict:
        return {"capture_id": coord.start(host=host, adapter=adapter, note=note)}

    @mcp.tool()
    def jdsl_capture_finish(capture_id: str) -> dict:
        coord.finish(capture_id)
        return {"ok": True}

    @mcp.tool()
    def jdsl_capture_mark_outcome(capture_id: str, episode_id: str, reward: float | None = None,
                                  verdict: str | None = None) -> dict:
        coord.mark_outcome(capture_id, episode_id, reward=reward, verdict=verdict)
        return {"ok": True}

    @mcp.tool()
    def jdsl_capture_summary(capture_id: str) -> dict:
        return coord.summary(capture_id)

    @mcp.tool()
    def jdsl_inspect(capture_id: str) -> dict:
        return coord.lineage_report(capture_id)

    @mcp.tool()
    def jdsl_compile(capture_id: str, name: str = "behavior") -> dict:
        from jdsl_harness.compiler import compile_behavior
        result = compile_behavior(store.capture_episodes(capture_id), name=name)
        return result.report()

    return mcp

Compiler Passes

normalize

Deterministic trajectory normalization (design §13, §35 PR8).

Raw traces carry surface variation; the compiler needs symbolic structure. This stage turns a canonical Episode into a NormEpisode: an ordered list of action steps whose arguments are annotated with exact dataflow lineage (§16.1) and whose instance-specific values are replaced with symbolic references (§13.2). No model is used — every decision here is structural and reproducible.

Classes:

• NormStep –

  One normalized action step (a tool call) with lineage-annotated arguments.

• ModelDecision –

  A residual semantic decision observed in the trace (react/predict leaf).

• NormEpisode –

Functions:

• normalize_episode –

  Normalize one episode. canonical maps host tool names to logical ids

• normalize_all –
• synth_store –

  The store name assigned to a tool result that had no explicit store. Shared

• synth_node_id –

NormStep dataclass

NormStep(index: int, logical_tool: str, host_tool: str | None, arguments: dict[str, Any], arg_lineage: dict[str, str | None], result: Any = None, ok: bool = True, error: Any = None, store: str | None = None, node_id: str | None = None, state_before: dict[str, Any] = dict())

One normalized action step (a tool call) with lineage-annotated arguments.

Methods:

• symbolic_args –

  Arguments with lineaged values replaced by $ref(path) markers (§13.2).

symbolic_args

symbolic_args() -> dict[str, Any]

Arguments with lineaged values replaced by $ref(path) markers (§13.2).

Source code in jdsl_harness/compiler/normalize.py

def symbolic_args(self) -> dict[str, Any]:
    """Arguments with lineaged values replaced by `$ref(path)` markers (§13.2)."""
    out: dict[str, Any] = {}
    for k, v in self.arguments.items():
        path = self.arg_lineage.get(k)
        out[k] = {"ref": path} if path else {"const": v}
    return out

ModelDecision dataclass

ModelDecision(index: int, node_id: str | None, inputs: list[str], outputs: list[str], kind: str = 'predict')

A residual semantic decision observed in the trace (react/predict leaf).

NormEpisode dataclass

NormEpisode(episode_id: str, steps: list[NormStep] = list(), decisions: list[ModelDecision] = list(), success: bool | None = None, outcome: dict[str, Any] | None = None, source_digest: str | None = None)

normalize_episode

normalize_episode(episode: Episode, *, canonical: dict[str, str] | None = None) -> NormEpisode

Normalize one episode. canonical maps host tool names to logical ids (§13.1 tool canonicalization); unmapped tools keep their host name.

Source code in jdsl_harness/compiler/normalize.py

def normalize_episode(episode: Episode, *, canonical: dict[str, str] | None = None) -> NormEpisode:
    """Normalize one episode. `canonical` maps host tool names to logical ids
    (§13.1 tool canonicalization); unmapped tools keep their host name."""
    canonical = canonical or {}
    norm = NormEpisode(episode_id=episode.episode_id)
    env: dict[str, Any] = {}          # cumulative trusted state before the current step
    pending: dict[str, NormStep] = {}  # started tool calls awaiting result, by event id
    index = 0

    for e in episode.ordered():
        if e.kind == EventKind.BLACKBOARD_WRITE:
            env[e.payload["key"]] = e.payload.get("value")
        elif e.kind == EventKind.TOOL_CALL_STARTED:
            tool = e.payload.get("tool", {})
            host = tool.get("host_name")
            logical = tool.get("logical_id") or canonical.get(host or "", host or "unknown")
            arguments = dict(e.payload.get("arguments", {}) or {})
            import copy
            lineage = {k: find_source(v, env) for k, v in arguments.items()}
            step = NormStep(index=index, logical_tool=logical, host_tool=host,
                            arguments=arguments, arg_lineage=lineage,
                            store=e.payload.get("store"), node_id=e.payload.get("node_id"),
                            state_before=copy.deepcopy(env))
            norm.steps.append(step)
            pending[e.event_id] = step
            index += 1
        elif e.kind == EventKind.TOOL_CALL_COMPLETED:
            step = pending.get(e.parent_event_id or "") or (norm.steps[-1] if norm.steps else None)
            if step is not None:
                step.result = e.payload.get("result")
                # register the result under its store name so later steps can
                # reference it (the symbolic $var of §13.2); also feed env. When the
                # trace carried no store name (gateway/imported calls), synthesize a
                # stable one the staticizer/verifier reuse.
                name = e.payload.get("store") or step.store or synth_store(step.logical_tool, step.index)
                step.store = name
                env[name] = step.result
        elif e.kind == EventKind.TOOL_CALL_FAILED:
            step = pending.get(e.parent_event_id or "") or (norm.steps[-1] if norm.steps else None)
            if step is not None:
                step.ok = False
                step.error = e.payload.get("error")
        elif e.kind == EventKind.REACT_STARTED:
            norm.decisions.append(ModelDecision(
                index=index, node_id=e.payload.get("node_id"),
                inputs=e.payload.get("inputs", []), outputs=e.payload.get("outputs", []), kind="react"))
        elif e.kind == EventKind.NODE_ENTER and e.payload.get("type") == "predict":
            ins, outs = _parse_predict_label(e.payload.get("label", ""))
            norm.decisions.append(ModelDecision(
                index=index, node_id=e.payload.get("node_id"), inputs=ins, outputs=outs, kind="predict"))
        elif e.kind in (EventKind.ENVIRONMENT_REWARD, EventKind.ENVIRONMENT_VERDICT):
            norm.outcome = e.payload

    norm.success = episode.succeeded()
    ordered = episode.ordered()
    norm.source_digest = ordered[-1].event_hash if ordered else None
    return norm

normalize_all

normalize_all(episodes: list[Episode], *, canonical: dict[str, str] | None = None) -> list[NormEpisode]

Source code in jdsl_harness/compiler/normalize.py

def normalize_all(episodes: list[Episode], *, canonical: dict[str, str] | None = None) -> list[NormEpisode]:
    return [normalize_episode(e, canonical=canonical) for e in episodes]

synth_store

synth_store(tool: str, index: int) -> str

The store name assigned to a tool result that had no explicit store. Shared by the normalizer, staticizer, and verifier so refs stay consistent (§13.2).

Source code in jdsl_harness/compiler/normalize.py

def synth_store(tool: str, index: int) -> str:
    """The store name assigned to a tool result that had no explicit store. Shared
    by the normalizer, staticizer, and verifier so refs stay consistent (§13.2)."""
    return f"{slug(tool)}_out_{index}"

synth_node_id

synth_node_id(tool: str, index: int) -> str

Source code in jdsl_harness/compiler/normalize.py

def synth_node_id(tool: str, index: int) -> str:
    return f"{slug(tool)}_{index}"

lineage

Exact dataflow lineage (design §16.1, §13.2).

The single most valuable deterministic signal: when a tool argument value is exactly a value that already exists in trusted state, the model should not regenerate it — it should reference it (§16.1 "If the identifier already exists in trusted state, the model should not regenerate the identifier"). This module finds, for a given value, the JSON path in a prior state that produced it. No model involved; this is pure structural comparison.

Functions:

• is_meaningful –

  Only mine lineage for identifier-like values: non-trivial scalars. Strings

• find_source –

  Return the best JSON path in state whose value equals value exactly, or

• find_all_sources –

  Every path in state equal to value. Ordered best-first: symbolic-index

is_meaningful

is_meaningful(value: Any) -> bool

Only mine lineage for identifier-like values: non-trivial scalars. Strings must be at least 2 chars; small ints/bools are too common to attribute.

Source code in jdsl_harness/compiler/lineage.py

def is_meaningful(value: Any) -> bool:
    """Only mine lineage for identifier-like values: non-trivial scalars. Strings
    must be at least 2 chars; small ints/bools are too common to attribute."""
    if value in _TRIVIAL:
        return False
    if isinstance(value, bool):
        return False
    if isinstance(value, str):
        return len(value) >= 2
    if isinstance(value, int):
        return abs(value) > 1
    if isinstance(value, float):
        return True
    return False

find_source

find_source(value: Any, state: dict[str, Any], *, max_depth: int = 6) -> str | None

Return the best JSON path in state whose value equals value exactly, or None. Paths use dotted keys and [i] indices (the ref syntax of §21.1). A list index equal to a state variable is emitted symbolically (orders[$selected_index], §49) so it generalizes to the model's decision rather than hardcoding the row.

Source code in jdsl_harness/compiler/lineage.py

def find_source(value: Any, state: dict[str, Any], *, max_depth: int = 6) -> str | None:
    """Return the best JSON path in `state` whose value equals `value` exactly, or
    None. Paths use dotted keys and `[i]` indices (the ref syntax of §21.1). A list
    index equal to a state variable is emitted symbolically (`orders[$selected_index]`,
    §49) so it generalizes to the model's decision rather than hardcoding the row."""
    if not is_meaningful(value):
        return None
    paths = find_all_sources(value, state, max_depth=max_depth)
    return paths[0] if paths else None

find_all_sources

find_all_sources(value: Any, state: dict[str, Any], *, max_depth: int = 6) -> list[str]

Every path in state equal to value. Ordered best-first: symbolic-index paths (which generalize) before literal-index paths, then shortest.

Source code in jdsl_harness/compiler/lineage.py

def find_all_sources(value: Any, state: dict[str, Any], *, max_depth: int = 6) -> list[str]:
    """Every path in `state` equal to `value`. Ordered best-first: symbolic-index
    paths (which generalize) before literal-index paths, then shortest."""
    if not is_meaningful(value):
        return []
    index_names = _index_names(state)
    found: list[str] = []
    for key, sub in state.items():
        found.extend(_search(sub, str(key), value, max_depth, index_names))
    # prefer generalizing ($) paths, then shorter ones; stable for determinism
    return sorted(set(found), key=lambda p: (p.count("$") == 0, len(p), p))

candidates

Behavior candidate mining — the six atom types (design §4.2, §14.1, §16, §35 PR9).

Stage A of the compiler (§14.1): analyze each normalized episode independently and extract local behavior facts — no generalization yet. Facts carry a grouping key so consolidate.py (Stage B) can measure support and counterexamples across many episodes and assign an evidence grade (§15).

Classes:

• Fact –

  One local behavior observation from a single episode.

Functions:

• extract_facts –

  All local facts from one normalized episode (§14.1).

Fact dataclass

Fact(type: str, claim: dict[str, Any], episode_id: str, outcome_ok: bool | None = None, state_before: dict[str, Any] = dict())

One local behavior observation from a single episode.

Methods:

• key –

  Stable identity for grouping equivalent facts across episodes.

key

key() -> tuple

Stable identity for grouping equivalent facts across episodes.

Source code in jdsl_harness/compiler/candidates.py

def key(self) -> tuple:
    """Stable identity for grouping equivalent facts across episodes."""
    return (self.type, _freeze(self.claim))

extract_facts

extract_facts(ep: NormEpisode) -> list[Fact]

All local facts from one normalized episode (§14.1).

Source code in jdsl_harness/compiler/candidates.py

def extract_facts(ep: NormEpisode) -> list[Fact]:
    """All local facts from one normalized episode (§14.1)."""
    facts: list[Fact] = []
    facts += _dataflow_facts(ep)
    facts += _control_facts(ep)
    facts += _action_facts(ep)
    facts += _recovery_facts(ep)
    facts += _semantic_facts(ep)
    return facts

consolidate

Cross-trace consolidation and evidence grading (design §14.2, §15, §35 PR9).

Stage B of the compiler: group equivalent local facts, measure support and counterexamples against the episodes where each claim was applicable, and assign a conservative evidence grade (§15). Frequency alone is never enough — a claim contradicted in an applicable episode is contested and cannot become a hard rule (§15, §44.2). Grades E4/E5 are reserved for the verifier and held-out evaluation.

Classes:

• Evidence –
• Candidate –

Functions:

• consolidate –

  Consolidate local facts from all episodes into graded candidates. contracts

Evidence dataclass

Evidence(applicable: int = 0, support: int = 0, counterexamples: int = 0, episodes: list[str] = list(), success_support: int = 0, fail_support: int = 0)

Candidate dataclass

Candidate(candidate_id: str, type: str, claim: dict[str, Any], evidence: Evidence, grade: str = E0, status: str = 'proposed', contract_sources: list[str] = list())

consolidate

consolidate(episodes: list[NormEpisode], *, contracts: dict[tuple, list[str]] | None = None) -> list[Candidate]

Consolidate local facts from all episodes into graded candidates. contracts optionally maps a fact key to contract source ids, lifting it to E3 (§15 E3).

Source code in jdsl_harness/compiler/consolidate.py

def consolidate(episodes: list[NormEpisode], *,
                contracts: dict[tuple, list[str]] | None = None) -> list[Candidate]:
    """Consolidate local facts from all episodes into graded candidates. `contracts`
    optionally maps a fact key to contract source ids, lifting it to E3 (§15 E3)."""
    contracts = contracts or {}
    grouped: dict[tuple, list[Fact]] = {}
    for ep in episodes:
        seen_keys: set[tuple] = set()
        for fact in extract_facts(ep):
            k = fact.key()
            # count each distinct claim at most once per episode for support
            if k in seen_keys:
                continue
            seen_keys.add(k)
            grouped.setdefault(k, []).append(fact)

    candidates: list[Candidate] = []
    for i, (key, facts) in enumerate(sorted(grouped.items(), key=lambda kv: str(kv[0]))):
        rep = facts[0]
        ev = _evidence(rep, facts, episodes)
        cand = Candidate(candidate_id=f"cand_{rep.type.lower()}_{i:03d}", type=rep.type,
                         claim=rep.claim, evidence=ev)
        cand.contract_sources = contracts.get(key, [])
        cand.grade, cand.status = _grade(ev, bool(cand.contract_sources))
        candidates.append(cand)
    return candidates

staticize

Staticization: build the Behavior IR from evidence (design §17, §25, §35 PR11).

The compiler objective is lowest model burden while preserving verified behavior (§17, §2.3). We follow the staticization ordering (§17): constant? exact dataflow? deterministic predicate? fixed action? bounded recovery? classification? — and only what survives as genuine judgment stays a residual model leaf.

The structural method is §25: build a control skeleton from the modal successful trajectory, replace instance values with refs, insert verified guards, wrap known recovery, and residualize the rest.

Classes:

• CompiledBehavior –

  The staticizer's output: an IR, its provenance, and burden metrics (§33).

Functions:

• staticize –

CompiledBehavior dataclass

CompiledBehavior(ir: BehaviorIR, provenance: list[NodeProvenance] = list(), stats: dict[str, Any] = dict(), required_capabilities: list[str] = list())

The staticizer's output: an IR, its provenance, and burden metrics (§33).

staticize

staticize(episodes: list[NormEpisode], candidates: list[Candidate], *, name: str = 'behavior', model: CompilerModel | None = None, compiler_model_id: str | None = None) -> CompiledBehavior

Source code in jdsl_harness/compiler/staticize.py

def staticize(episodes: list[NormEpisode], candidates: list[Candidate], *,
              name: str = "behavior", model: CompilerModel | None = None,
              compiler_model_id: str | None = None) -> CompiledBehavior:
    model = model or HeuristicCompilerModel()
    skeleton = _pick_skeleton(episodes)
    if skeleton is None:
        empty = BehaviorIR(root=IRSequence(type="sequence", id=f"{name}_flow", children_=[]))
        return CompiledBehavior(ir=empty, stats={"note": "no successful episode to compile"})

    dataflow = _index_dataflow(candidates)
    recovery = _index_recovery(candidates)
    guards = _index_guards(candidates)
    inputs = _input_args(episodes, dataflow)

    children: list[IRNode] = []
    signatures: dict[str, Signature] = {}
    provenance: list[NodeProvenance] = []
    caps: set[str] = set()
    decisions_by_index = _decisions_by_index(skeleton)

    n_deterministic = 0
    n_model = 0

    for i, step in enumerate(skeleton.steps):
        # interleave residual decisions observed just before this step
        for decision in decisions_by_index.get(i, []):
            sig, leaf = residualize_decision(decision, model=model)
            signatures[sig.id] = sig
            children.append(leaf)
            provenance.append(NodeProvenance(node_id=leaf.id or sig.id, evidence_grade="E0",
                                             behavior_candidate=None, compiler_model=compiler_model_id,
                                             rationale_summary="residual semantic decision"))
            n_model += 1
            caps.update(sig.tools)

        # a verified guard scoped before this tool (§16.3)
        for gcand in guards.get(step.logical_tool, []):
            gnode = IRGuard(type="guard", id=f"guard_{step.logical_tool}_{i}".replace(".", "_"),
                            expression=gcand.claim["expression"])
            children.append(gnode)
            provenance.append(_prov(gnode.id, gcand, compiler_model_id))
            n_deterministic += 1

        action = _build_action(step, i, dataflow, inputs, caps)
        acand = _action_candidate(candidates, step)
        prov = _prov(action.id, acand, compiler_model_id) if acand else NodeProvenance(
            node_id=action.id or "", evidence_grade="E0", compiler_model=compiler_model_id)

        # known bounded recovery -> sel(action, recovery) (§16.5)
        rcand = recovery.get(step.logical_tool)
        if rcand is not None:
            rec_tool = rcand.claim["recover_with"]
            caps.add(rec_tool)
            recover = IRAction(type="action", id=f"recover_{i}", tool=rec_tool,
                               store=action.store)
            node: IRNode = IRSelector(type="selector", id=f"try_{action.id}",
                                      children_=[action, recover])
            provenance.append(_prov(node.id, rcand, compiler_model_id))
        else:
            node = action
        provenance.append(prov)
        children.append(node)
        n_deterministic += 1

    # any trailing decisions after the last step
    for decision in decisions_by_index.get(len(skeleton.steps), []):
        sig, leaf = residualize_decision(decision, model=model)
        signatures[sig.id] = sig
        children.append(leaf)
        provenance.append(NodeProvenance(node_id=leaf.id or sig.id, evidence_grade="E0",
                                         compiler_model=compiler_model_id,
                                         rationale_summary="residual semantic decision"))
        n_model += 1
        caps.update(sig.tools)

    root = IRSequence(type="sequence", id=f"{name}_flow", children_=children)
    ir = BehaviorIR(root=root, signatures=signatures)
    total = n_deterministic + n_model
    declared_inputs = sorted(set(inputs.values()))
    stats = {
        "meaningful_decisions": total,
        "model_dependent_decisions": n_model,
        "residual_decision_burden": round(n_model / total, 4) if total else 0.0,
        "deterministic_coverage": round(n_deterministic / total, 4) if total else 0.0,
        "exact_dataflow_refs": sum(1 for s in skeleton.steps for v in s.arg_lineage.values() if v),
        "inputs": declared_inputs,
    }
    return CompiledBehavior(ir=ir, provenance=provenance, stats=stats,
                            required_capabilities=sorted(caps))

residualize

Residualize the semantic decisions the compiler could not remove (design §18, §26, §35 PR10).

Everything the staticizer can prove deterministic has been lowered to control, dataflow, guards, and fixed actions. What remains — genuine language judgment — is turned into typed residual signatures (§18): the small interface between one tree leaf and the frozen model. A residual leaf prefers predict + deterministic act over a wide-open react (§18.3).

Functions:

• residualize_decision –

  Turn one observed model decision into a typed Signature + its IR leaf.

residualize_decision

residualize_decision(decision: ModelDecision, *, model: CompilerModel | None = None) -> tuple[Signature, IRPredict | IRReact]

Turn one observed model decision into a typed Signature + its IR leaf.

Source code in jdsl_harness/compiler/residualize.py

def residualize_decision(decision: ModelDecision, *, model: CompilerModel | None = None,
                         ) -> tuple[Signature, IRPredict | IRReact]:
    """Turn one observed model decision into a typed Signature + its IR leaf."""
    model = model or HeuristicCompilerModel()
    payload = {"node_id": decision.node_id, "inputs": decision.inputs,
               "outputs": decision.outputs, "kind": decision.kind}
    sig = model.name_signature(payload)
    node_id = decision.node_id or sig.id
    if sig.kind == "react":
        return sig, IRReact(type="react", id=node_id, signature=sig.id)
    return sig, IRPredict(type="predict", id=node_id, signature=sig.id)

verify

Verification: deterministic layers that decide what survives (design §32, §35 PR12).

Compilation without verification produces fragile policies (§32). None of this trusts the compiler model — every check is deterministic (§24.1): structural validity (§32.1), then replay of the compiled deterministic behavior against the historical episodes (§32.2). Dataflow refs that reproduce the recorded values, and guards that pick the recorded branch, are promoted to replay-verified (E4).

Classes:

• VerificationReport –

Functions:

• verify –
• promote_replay_verified –

  When replay is fully clean, lift each compiled node's provenance to E4

VerificationReport dataclass

VerificationReport(structural_ok: bool = True, structural_problems: list[str] = list(), replay_checks: int = 0, replay_passed: int = 0, problems: list[str] = list())

verify

verify(compiled: CompiledBehavior, episodes: list[NormEpisode], *, required_capabilities: set[str] | None = None) -> VerificationReport

Source code in jdsl_harness/compiler/verify.py

def verify(compiled: CompiledBehavior, episodes: list[NormEpisode], *,
           required_capabilities: set[str] | None = None) -> VerificationReport:
    report = VerificationReport()

    structural = validate_ir(compiled.ir, required_capabilities=required_capabilities)
    report.structural_ok = structural.ok
    report.structural_problems = list(structural.problems)

    _replay(compiled, episodes, report)
    return report

promote_replay_verified

promote_replay_verified(compiled: CompiledBehavior, report: VerificationReport) -> None

When replay is fully clean, lift each compiled node's provenance to E4 (replay-verified, §15). Safety-sensitive guards still needed E3 to be compiled.

Source code in jdsl_harness/compiler/verify.py

def promote_replay_verified(compiled: CompiledBehavior, report: VerificationReport) -> None:
    """When replay is fully clean, lift each compiled node's provenance to E4
    (replay-verified, §15). Safety-sensitive guards still needed E3 to be compiled."""
    if report.ok and report.replay_checks > 0:
        for prov in compiled.provenance:
            if prov.evidence_grade in ("E1", "E2"):
                prov.evidence_grade = E4

package

Assemble a portable BehaviorPackage from compiled + verified behavior (design §22, §23, §35 PR13) and the end-to-end compile pipeline (§24).

compile_behavior is the whole spine (§24): normalize → consolidate → staticize → verify → package. It is deterministic given the traces and the (optional) compiler model; the model only proposes wording/signatures, never the structure or the evidence counts (§24.1).

Classes:

• CompileResult –

  The full output of one compile run — package plus every intermediate.

Functions:

• compile_behavior –

  Run the whole compiler pipeline over canonical trace episodes (§24).

• build_package –

CompileResult

CompileResult(*, package: BehaviorPackage, compiled: CompiledBehavior, candidates: list[Candidate], verification: VerificationReport, normalized: list[NormEpisode])

The full output of one compile run — package plus every intermediate.

Source code in jdsl_harness/compiler/package.py

def __init__(self, *, package: BehaviorPackage, compiled: CompiledBehavior,
             candidates: list[Candidate], verification: VerificationReport,
             normalized: list[NormEpisode]) -> None:
    self.package = package
    self.compiled = compiled
    self.candidates = candidates
    self.verification = verification
    self.normalized = normalized

compile_behavior

compile_behavior(episodes: list[Episode], *, name: str = 'behavior', task_family: str = '', canonical: dict[str, str] | None = None, contracts: dict[tuple, list[str]] | None = None, effects: dict[str, ToolEffects] | None = None, model: CompilerModel | None = None, compiler_model_id: str | None = None, capture_fidelity: str = 'F3', version: str = '0.1.0') -> CompileResult

Run the whole compiler pipeline over canonical trace episodes (§24).

Source code in jdsl_harness/compiler/package.py

def compile_behavior(episodes: list[Episode], *, name: str = "behavior",
                     task_family: str = "", canonical: dict[str, str] | None = None,
                     contracts: dict[tuple, list[str]] | None = None,
                     effects: dict[str, ToolEffects] | None = None,
                     model: CompilerModel | None = None, compiler_model_id: str | None = None,
                     capture_fidelity: str = "F3", version: str = "0.1.0") -> CompileResult:
    """Run the whole compiler pipeline over canonical trace episodes (§24)."""
    norm = normalize_all(episodes, canonical=canonical)
    candidates = consolidate(norm, contracts=contracts)
    compiled = staticize(norm, candidates, name=name, model=model, compiler_model_id=compiler_model_id)
    report = verify(compiled, norm, required_capabilities=set(compiled.required_capabilities))
    promote_replay_verified(compiled, report)
    pkg = build_package(compiled, report, candidates, norm, name=name, task_family=task_family,
                        effects=effects, compiler_model_id=compiler_model_id,
                        capture_fidelity=capture_fidelity, version=version)
    return CompileResult(package=pkg, compiled=compiled, candidates=candidates,
                         verification=report, normalized=norm)

build_package

build_package(compiled: CompiledBehavior, report: VerificationReport, candidates: list[Candidate], norm: list[NormEpisode], *, name: str, task_family: str = '', effects: dict[str, ToolEffects] | None = None, compiler_model_id: str | None = None, capture_fidelity: str = 'F3', version: str = '0.1.0') -> BehaviorPackage

Source code in jdsl_harness/compiler/package.py

def build_package(compiled: CompiledBehavior, report: VerificationReport,
                  candidates: list[Candidate], norm: list[NormEpisode], *,
                  name: str, task_family: str = "", effects: dict[str, ToolEffects] | None = None,
                  compiler_model_id: str | None = None, capture_fidelity: str = "F3",
                  version: str = "0.1.0") -> BehaviorPackage:
    effects = effects or {}
    caps = compiled.required_capabilities
    tools = [ToolContract(logical_id=c, effects=effects.get(c, _default_effects(c))) for c in caps]

    manifest = Manifest(
        name=name, version=version, task_family=task_family, required_capabilities=list(caps),
        source={"compiler": "jdsl-compiler", "compiler_model": compiler_model_id,
                "capture_fidelity": capture_fidelity, "episode_count": len(norm),
                "inputs": compiled.stats.get("inputs", [])},
        verification=report.to_dict(),
    )
    evidence_summary = {
        "episode_count": len(norm),
        "successful": sum(1 for e in norm if e.success),
        "candidates_by_grade": _by_grade(candidates),
        "candidates_by_type": _by_type(candidates),
        "residual_decision_burden": compiled.stats.get("residual_decision_burden"),
        "deterministic_coverage": compiled.stats.get("deterministic_coverage"),
    }
    tests = {
        "replay": [{"episode": e.episode_id, "success": e.success,
                    "source_digest": e.source_digest} for e in norm],
        "guards": [c.to_dict() for c in candidates if c.type == "GUARD"],
        "signatures": [{"id": sid, **sig.to_dict()} for sid, sig in compiled.ir.signatures.items()],
    }
    return BehaviorPackage(
        manifest=manifest, ir=compiled.ir, tools=tools, provenance=compiled.provenance,
        tests=tests, evidence_summary=evidence_summary,
        invariants=[c.to_dict() for c in candidates if c.grade == "E3" and c.type == "GUARD"],
    )

Host Adapters

correlation

Tool-call correlation for host hooks.

Host integrations often deliver "before tool" and "after tool" as separate HTTP posts. This small state object remembers started calls by host-provided call id and annotates completion events with the correct parent_event_id so replay and normalization do not have to guess from sequence order.

Classes:

• ToolCallCorrelator –

  Short-lived in-memory correlation state scoped to one ingest process.

Functions:

• host_call_id –

  Common field names used by host hook payloads for a tool-call id.

ToolCallCorrelator dataclass

ToolCallCorrelator(_by_id: dict[tuple[str, str, str], str] = dict(), _open: dict[tuple[str, str], list[tuple[str, str, str | None]]] = dict(), _counter: Any = (lambda: itertools.count(1))())

Short-lived in-memory correlation state scoped to one ingest process.

host_call_id

host_call_id(payload: dict[str, Any]) -> str | None

Common field names used by host hook payloads for a tool-call id.

Source code in jdsl_harness/adapters/correlation.py

def host_call_id(payload: dict[str, Any]) -> str | None:
    """Common field names used by host hook payloads for a tool-call id."""
    for key in ("call_id", "tool_call_id", "tool_use_id", "invocation_id", "id"):
        value = payload.get(key)
        if value not in (None, ""):
            return str(value)
    return None

claude_code

Claude Code host adapter (design §8.2, §29.1).

Translates Claude Code's structured hook payloads into canonical jdsl events. The adapter uses only the structured JSON hook payload — never scraped terminal text (§8.2). It is a pure function of the payload so it can be unit-tested against recorded fixtures without running the host (§46 hook fixture tests).

Claude Code hook events used (§29.1): SessionStart, UserPromptSubmit, PreToolUse, PostToolUse, PostToolUseFailure, SessionEnd. The tool call is correlated across Pre/Post by (session, tool_name, tool_input).

Functions:

• to_events –

  Map one Claude Code hook payload to zero or more canonical events. The

to_events

to_events(payload: dict[str, Any], *, capture_id: str, model: str | None = None, correlator: ToolCallCorrelator | None = None) -> list[TraceEvent]

Map one Claude Code hook payload to zero or more canonical events. The session id becomes the episode id; unknown hook names are dropped (capture fidelity is recorded elsewhere, §8.2).

Source code in jdsl_harness/adapters/claude_code.py

def to_events(payload: dict[str, Any], *, capture_id: str, model: str | None = None,
              correlator: ToolCallCorrelator | None = None) -> list[TraceEvent]:
    """Map one Claude Code hook payload to zero or more canonical events. The
    session id becomes the episode id; unknown hook names are dropped (capture
    fidelity is recorded elsewhere, §8.2)."""
    hook = payload.get("hook_event_name") or payload.get("hook")
    episode_id = payload.get("session_id") or "ep_claude"
    src = EventSource(host=HOST, adapter=ADAPTER, model=model)

    def ev(kind: str, *, actor: str = "system", data: dict[str, Any] | None = None) -> TraceEvent:
        return TraceEvent.new(kind, capture_id, episode_id, payload=data or {}, actor=actor, source=src)

    if hook == "SessionStart":
        return [ev(EventKind.EPISODE_STARTED, data={"host": HOST})]
    if hook == "UserPromptSubmit":
        return [ev(EventKind.USER_MESSAGE, actor="user",
                   data={"text": payload.get("prompt") or payload.get("user_prompt", "")})]
    if hook == "PreToolUse":
        event = ev(EventKind.TOOL_CALL_STARTED, actor="model", data={
            "tool": {"host_name": payload.get("tool_name"), "logical_id": None},
            "arguments": payload.get("tool_input", {})})
        cid = host_call_id(payload)
        if correlator is not None:
            event = correlator.started(event, host_call_id=cid, tool_name=payload.get("tool_name"))
        elif cid is not None:
            event.payload["host_call_id"] = cid
        return [event]
    if hook == "PostToolUse":
        response = payload.get("tool_response", payload.get("tool_result"))
        event = ev(EventKind.TOOL_CALL_COMPLETED, actor="tool", data={
            "tool": {"host_name": payload.get("tool_name")},
            "result": _normalize_result(response)})
        cid = host_call_id(payload)
        if correlator is not None:
            event = correlator.finished(event, host_call_id=cid, tool_name=payload.get("tool_name"))
        elif cid is not None:
            event.payload["host_call_id"] = cid
        return [event]
    if hook in ("PostToolUseFailure", "PostToolBatchFailure"):
        event = ev(EventKind.TOOL_CALL_FAILED, actor="tool", data={
            "tool": {"host_name": payload.get("tool_name")},
            "error": payload.get("error") or payload.get("tool_response")})
        cid = host_call_id(payload)
        if correlator is not None:
            event = correlator.finished(event, host_call_id=cid, tool_name=payload.get("tool_name"))
        elif cid is not None:
            event.payload["host_call_id"] = cid
        return [event]
    if hook in ("SubagentStart", "SubagentStop"):
        kind = EventKind.HOST_SUBAGENT_STARTED if hook == "SubagentStart" else EventKind.HOST_SUBAGENT_FINISHED
        return [ev(kind, data={"agent": payload.get("subagent_type")})]
    if hook == "SessionEnd":
        return [ev(EventKind.EPISODE_FINISHED, data={"host": HOST})]
    return []

gemini_cli

Gemini CLI host adapter (design §8.2, §29.2).

Maps Gemini CLI's hook payloads to canonical jdsl events. Gemini exposes a broader surface than Claude Code, including model and tool-selection events (§29.2); for the first release these are used for capture only (not enforcement). Full model requests are not stored by default (§29.2).

Hooks used: SessionStart, BeforeAgent, BeforeToolSelection, BeforeTool, AfterTool, AfterAgent, SessionEnd.

Functions:

• to_events –

to_events

to_events(payload: dict[str, Any], *, capture_id: str, model: str | None = None, correlator: ToolCallCorrelator | None = None) -> list[TraceEvent]

Source code in jdsl_harness/adapters/gemini_cli.py

def to_events(payload: dict[str, Any], *, capture_id: str, model: str | None = None,
              correlator: ToolCallCorrelator | None = None) -> list[TraceEvent]:
    hook = payload.get("hook") or payload.get("event")
    episode_id = payload.get("session_id") or payload.get("sessionId") or "ep_gemini"
    src = EventSource(host=HOST, adapter=ADAPTER, model=model)

    def ev(kind: str, *, actor: str = "system", data: dict[str, Any] | None = None) -> TraceEvent:
        return TraceEvent.new(kind, capture_id, episode_id, payload=data or {}, actor=actor, source=src)

    if hook in ("SessionStart", "BeforeAgent"):
        return [ev(EventKind.EPISODE_STARTED, data={"host": HOST})] if hook == "SessionStart" else []
    if hook == "BeforeToolSelection":
        tools = payload.get("available_tools") or payload.get("tools") or []
        return [ev(EventKind.TOOLSET_EXPOSED, data={
            "tools": [{"host_name": t if isinstance(t, str) else t.get("name")} for t in tools]})]
    if hook == "BeforeTool":
        tool = payload.get("tool_name") or payload.get("name")
        event = ev(EventKind.TOOL_CALL_STARTED, actor="model", data={
            "tool": {"host_name": payload.get("tool_name") or payload.get("name")},
            "arguments": payload.get("args") or payload.get("tool_input", {})})
        cid = host_call_id(payload)
        if correlator is not None:
            event = correlator.started(event, host_call_id=cid, tool_name=tool)
        elif cid is not None:
            event.payload["host_call_id"] = cid
        return [event]
    if hook == "AfterTool":
        tool = payload.get("tool_name") or payload.get("name")
        cid = host_call_id(payload)
        error = payload.get("error")
        if error:
            event = ev(EventKind.TOOL_CALL_FAILED, actor="tool", data={
                "tool": {"host_name": tool}, "error": error})
        else:
            event = ev(EventKind.TOOL_CALL_COMPLETED, actor="tool", data={
                "tool": {"host_name": tool},
                "result": payload.get("result") or payload.get("output")})
        if correlator is not None:
            event = correlator.finished(event, host_call_id=cid, tool_name=tool)
        elif cid is not None:
            event.payload["host_call_id"] = cid
        return [event]
    if hook in ("SessionEnd", "AfterAgent"):
        return [ev(EventKind.EPISODE_FINISHED, data={"host": HOST})] if hook == "SessionEnd" else []
    return []

opencode

OpenCode host adapter.

The TypeScript plugin translates OpenCode hook inputs into the stable jdsl.opencode-hook.v1 envelope. This module maps that envelope into canonical jdsl trace events without depending on OpenCode's TypeScript types.

Classes:

• OpenCodeEnvelopeError –

  An OpenCode hook payload does not match the jdsl stable envelope.

Functions:

• to_events –

  Map one stable OpenCode envelope to canonical trace events.

OpenCodeEnvelopeError

Bases: ValueError

An OpenCode hook payload does not match the jdsl stable envelope.

to_events

to_events(payload: dict[str, Any], *, capture_id: str, model: str | None = None, correlator: ToolCallCorrelator | None = None) -> list[TraceEvent]

Map one stable OpenCode envelope to canonical trace events.

Source code in jdsl_harness/adapters/opencode.py

def to_events(payload: dict[str, Any], *, capture_id: str, model: str | None = None,
              correlator: ToolCallCorrelator | None = None) -> list[TraceEvent]:
    """Map one stable OpenCode envelope to canonical trace events."""
    _validate_envelope(payload)
    hook = payload["hook"]
    episode_id = str(payload.get("episode_id") or payload["session_id"])
    src = EventSource(host=HOST, adapter=ADAPTER, model=model)

    def ev(kind: str, *, actor: str = "system", data: dict[str, Any] | None = None) -> TraceEvent:
        return TraceEvent.new(kind, capture_id, episode_id, payload=data or {}, actor=actor, source=src)

    if hook == "session.created":
        return [ev(EventKind.EPISODE_STARTED, data={
            "host": HOST,
            "session_id": payload["session_id"],
            "directory": payload.get("directory"),
            "worktree": payload.get("worktree"),
        })]
    if hook == "tool.execute.before":
        event = ev(EventKind.TOOL_CALL_STARTED, actor="model", data={
            "tool": {"host_name": payload.get("tool"), "logical_id": None},
            "arguments": payload.get("args") or {},
            "directory": payload.get("directory"),
            "worktree": payload.get("worktree"),
        })
        cid = host_call_id(payload)
        if correlator is not None:
            event = correlator.started(event, host_call_id=cid, tool_name=payload.get("tool"))
        elif cid is not None:
            event.payload["host_call_id"] = cid
        return [event]
    if hook == "tool.execute.after":
        error = payload.get("error")
        kind = EventKind.TOOL_CALL_FAILED if error else EventKind.TOOL_CALL_COMPLETED
        body = {
            "tool": {"host_name": payload.get("tool")},
            "directory": payload.get("directory"),
            "worktree": payload.get("worktree"),
        }
        if error:
            body["error"] = str(error)
        else:
            body["result"] = payload.get("result")
        event = ev(kind, actor="tool", data=body)
        cid = host_call_id(payload)
        if correlator is not None:
            event = correlator.finished(event, host_call_id=cid, tool_name=payload.get("tool"))
        elif cid is not None:
            event.payload["host_call_id"] = cid
        return [event]
    if hook == "session.error":
        return [ev(EventKind.ANNOTATION, data={
            "host": HOST,
            "session_id": payload["session_id"],
            "kind": "session.error",
            "error": payload.get("error"),
        })]
    if hook in ("session.deleted", "session.finished", "session.ended"):
        return [ev(EventKind.EPISODE_FINISHED, data={"host": HOST, "session_id": payload["session_id"]})]
    if hook in ("session.idle", "session.updated", "session.status", "session.compacted"):
        return [ev(EventKind.ANNOTATION, data={
            "host": HOST,
            "session_id": payload["session_id"],
            "kind": hook,
            "status": payload.get("status"),
        })]
    return []

mcp_proxy

Transparent MCP proxy — Tier-A capture for MCP-native tools (design §8.1.1, §35 PR5).

Many modern tools already arrive through MCP. Rather than writing a wrapper per tool, jdsl proxies an upstream MCP server: it discovers the upstream tools, preserves their input/output schemas, exposes them namespaced to the host, forwards calls, and records the full call + result to the trace store (§8.1.1).

The MCP SDK is an optional dependency (§36): this module defines the recording logic in a transport-neutral way (ProxiedTool, record_proxied_call) that works without mcp, and a serve_proxy entry point that lazily imports the SDK.

Classes:

• ProxiedTool –

  An upstream MCP tool the proxy exposes, with its preserved schema (§8.1.1).

• MCPProxy –

  Holds proxy configuration and the discovered upstream tool table. The actual

• StdioUpstream –

  One upstream MCP server reached over stdio.

Functions:

• record_proxied_call –

  Record one forwarded MCP call as canonical events (§8.1.1 items 4-6).

• build_stdio_proxy_server –

  Build a low-level MCP stdio proxy server for one upstream.

• discover_stdio_tools –

  Discover tools from one stdio upstream using the installed MCP SDK.

• call_stdio_tool –

  Forward one tool call to a stdio upstream.

• serve_proxy –

  Serve the stdio MCP proxy on this process's stdin/stdout.

ProxiedTool dataclass

ProxiedTool(server: str, name: str, input_schema: dict[str, Any] = dict(), output_schema: dict[str, Any] = dict(), description: str = '')

An upstream MCP tool the proxy exposes, with its preserved schema (§8.1.1).

Attributes:

• namespaced (str) –

  Host-visible name, namespaced by server to keep logical ids collision-free.

namespaced property

namespaced: str

Host-visible name, namespaced by server to keep logical ids collision-free.

MCPProxy dataclass

MCPProxy(sink: TraceSink, capture_id: str, episode_id: str = 'ep_proxy', tools: dict[str, ProxiedTool] = dict())

Holds proxy configuration and the discovered upstream tool table. The actual stdio/HTTP transport is provided by serve_proxy (needs the mcp SDK).

Methods:

• toolset_event –

  Emit the exposed toolset for the compiler's tool-visibility mining (§42).

toolset_event

toolset_event() -> TraceEvent

Emit the exposed toolset for the compiler's tool-visibility mining (§42).

Source code in jdsl_harness/mcp_proxy.py

def toolset_event(self) -> TraceEvent:
    """Emit the exposed toolset for the compiler's tool-visibility mining (§42)."""
    event = TraceEvent.new(EventKind.TOOLSET_EXPOSED, self.capture_id, self.episode_id,
                           source=EventSource(host="mcp-proxy", adapter="mcp-proxy"),
                           payload={"tools": [{"host_name": t.namespaced, "logical_id": t.logical_id,
                                               "description": t.description}
                                              for t in self.tools.values()]})
    return self.sink.emit(event)

StdioUpstream dataclass

StdioUpstream(server: str, command: str, args: list[str] = list(), env: dict[str, str] | None = None, cwd: str | Path | None = None)

One upstream MCP server reached over stdio.

record_proxied_call

record_proxied_call(sink: TraceSink, tool: ProxiedTool, arguments: dict[str, Any], *, capture_id: str, episode_id: str, result: Any = None, error: Any = None, host_call_id: str | None = None) -> list[TraceEvent]

Record one forwarded MCP call as canonical events (§8.1.1 items 4-6).

Source code in jdsl_harness/mcp_proxy.py

def record_proxied_call(sink: TraceSink, tool: ProxiedTool, arguments: dict[str, Any],
                        *, capture_id: str, episode_id: str, result: Any = None,
                        error: Any = None, host_call_id: str | None = None) -> list[TraceEvent]:
    """Record one forwarded MCP call as canonical events (§8.1.1 items 4-6)."""
    src = EventSource(host="mcp-proxy", adapter="mcp-proxy")
    start_payload = {
        "tool": {"host_name": tool.namespaced, "logical_id": tool.logical_id, "server": tool.server},
        "arguments": arguments,
    }
    if host_call_id is not None:
        start_payload["host_call_id"] = host_call_id
        start_payload["correlation"] = {"method": "host_call_id", "fidelity": "exact"}
    started = TraceEvent.new(EventKind.TOOL_CALL_STARTED, capture_id, episode_id, actor="model",
                             source=src, payload=start_payload)
    sink.emit(started)
    if error is not None:
        payload = {"tool": {"host_name": tool.namespaced}, "error": str(error)}
        if host_call_id is not None:
            payload["host_call_id"] = host_call_id
        done = TraceEvent.new(EventKind.TOOL_CALL_FAILED, capture_id, episode_id, actor="tool",
                              source=src, parent_event_id=started.event_id,
                              payload=payload)
    else:
        payload = {"tool": {"host_name": tool.namespaced}, "result": result}
        if host_call_id is not None:
            payload["host_call_id"] = host_call_id
        done = TraceEvent.new(EventKind.TOOL_CALL_COMPLETED, capture_id, episode_id, actor="tool",
                              source=src, parent_event_id=started.event_id,
                              payload=payload)
    sink.emit(done)
    return [started, done]

build_stdio_proxy_server

build_stdio_proxy_server(upstream: StdioUpstream | dict[str, Any], sink: TraceSink, *, capture_id: str, episode_id: str = 'ep_proxy', name: str = 'jdsl-mcp-proxy') -> Any

Build a low-level MCP stdio proxy server for one upstream.

The server discovers upstream tools, exposes namespaced copies with preserved JSON schemas, forwards calls to the upstream tool name, and records canonical tool-call events.

Source code in jdsl_harness/mcp_proxy.py

def build_stdio_proxy_server(upstream: StdioUpstream | dict[str, Any], sink: TraceSink, *,
                             capture_id: str, episode_id: str = "ep_proxy",
                             name: str = "jdsl-mcp-proxy") -> Any:
    """Build a low-level MCP stdio proxy server for one upstream.

    The server discovers upstream tools, exposes namespaced copies with preserved
    JSON schemas, forwards calls to the upstream tool name, and records canonical
    tool-call events.
    """
    try:
        from mcp import types
        from mcp.server.lowlevel import Server
    except ImportError as e:  # pragma: no cover - depends on optional extra
        raise RuntimeError("the MCP proxy transport needs the 'mcp' package: uv sync --extra harness") from e

    config = _upstream(upstream)
    proxy = MCPProxy(sink=sink, capture_id=capture_id, episode_id=episode_id)

    async def ensure_tools() -> None:
        if proxy.tools:
            return
        tools = await discover_stdio_tools(config)
        for tool in tools:
            proxy.register(ProxiedTool(
                server=config.server,
                name=tool.name,
                input_schema=tool.input_schema,
                output_schema=tool.output_schema or {},
                description=tool.description or "",
            ))
        proxy.toolset_event()

    async def on_list_tools(_ctx: Any, _params: Any) -> Any:
        await ensure_tools()
        return types.ListToolsResult(tools=[_as_mcp_tool(t) for t in proxy.tools.values()])

    async def on_call_tool(_ctx: Any, params: Any) -> Any:
        await ensure_tools()
        tool = proxy.tools.get(params.name)
        if tool is None:
            raise ValueError(f"unknown proxied tool {params.name!r}")
        arguments = params.arguments or {}
        result = await call_stdio_tool(config, tool.name, arguments)
        payload = _dump_mcp_result(result)
        if getattr(result, "is_error", False):
            proxy.record(tool.namespaced, arguments, error=payload)
        else:
            proxy.record(tool.namespaced, arguments, result=payload)
        return result

    server = Server(name, on_list_tools=on_list_tools, on_call_tool=on_call_tool)
    server._jdsl_on_list_tools = on_list_tools  # type: ignore[attr-defined]
    server._jdsl_on_call_tool = on_call_tool  # type: ignore[attr-defined]
    return server

discover_stdio_tools async

discover_stdio_tools(upstream: StdioUpstream | dict[str, Any]) -> list[Any]

Discover tools from one stdio upstream using the installed MCP SDK.

Source code in jdsl_harness/mcp_proxy.py

async def discover_stdio_tools(upstream: StdioUpstream | dict[str, Any]) -> list[Any]:
    """Discover tools from one stdio upstream using the installed MCP SDK."""
    config = _upstream(upstream)
    async with _stdio_session(config) as session:
        result = await session.list_tools()
        return list(result.tools)

call_stdio_tool async

call_stdio_tool(upstream: StdioUpstream | dict[str, Any], name: str, arguments: dict[str, Any]) -> Any

Forward one tool call to a stdio upstream.

Source code in jdsl_harness/mcp_proxy.py

async def call_stdio_tool(upstream: StdioUpstream | dict[str, Any], name: str,
                          arguments: dict[str, Any]) -> Any:
    """Forward one tool call to a stdio upstream."""
    config = _upstream(upstream)
    async with _stdio_session(config) as session:
        return await session.call_tool(name, arguments)

serve_proxy

serve_proxy(*, upstream: StdioUpstream | dict[str, Any], sink: TraceSink, capture_id: str, episode_id: str = 'ep_proxy', name: str = 'jdsl-mcp-proxy') -> None

Serve the stdio MCP proxy on this process's stdin/stdout.

Source code in jdsl_harness/mcp_proxy.py

def serve_proxy(*, upstream: StdioUpstream | dict[str, Any], sink: TraceSink,
                capture_id: str, episode_id: str = "ep_proxy",
                name: str = "jdsl-mcp-proxy") -> None:  # pragma: no cover - needs live MCP host
    """Serve the stdio MCP proxy on this process's stdin/stdout."""
    try:
        import anyio
        from mcp.server.stdio import stdio_server
    except ImportError as e:
        raise RuntimeError("the MCP proxy transport needs the 'mcp' package: uv sync --extra harness") from e

    async def run() -> None:
        server = build_stdio_proxy_server(upstream, sink, capture_id=capture_id,
                                          episode_id=episode_id, name=name)
        async with stdio_server() as (read_stream, write_stream):
            await server.run(read_stream, write_stream, server.create_initialization_options())

    anyio.run(run)