Skip to content

Trace and Package API¤

This page renders the tracing, IR, and package source APIs directly from code.

Trace Events¤

events ¤

The canonical, host-neutral trace event model (design §10).

Every observation the harness records — whether it came from the in-process jdsl runtime, a Claude Code hook, a Gemini CLI hook, an MCP proxy, or an imported log — is normalized into one TraceEvent envelope. Events are append-only and form a per-episode hash chain (§10.2) so a compiler can trust the provenance of the evidence it mines.

The design keeps the model deliberately small: a fixed set of kind strings (§10.1) over a free-form payload. Adapters emit the subset they can observe; nothing here assumes a particular host.

Classes:

  • EventKind

    The canonical event kinds (§10.1). A class of string constants rather than

  • EventSource

    Where an event came from (§10 source).

  • TraceEvent

    One canonical trace event. Construct via TraceEvent.new(...) so ids and

Attributes:

SCHEMA_VERSION module-attribute ¤

SCHEMA_VERSION = 'jdsl.trace.v1'

EventKind ¤

The canonical event kinds (§10.1). A class of string constants rather than an enum so importers can carry an unknown host kind through untouched while still comparing against the known set.

EventSource dataclass ¤

EventSource(host: str = 'jdsl', adapter: str = 'runtime', model: str | None = None)

Where an event came from (§10 source).

TraceEvent dataclass ¤

TraceEvent(kind: str, capture_id: str, episode_id: str, payload: dict[str, Any] = dict(), schema_version: str = SCHEMA_VERSION, event_id: str = '', sequence: int = 0, timestamp: str = '', source: EventSource = EventSource(), actor: str = 'system', parent_event_id: str | None = None, state_before_ref: str | None = None, state_after_ref: str | None = None, blob_refs: list[str] = list(), prev_event_hash: str | None = None, event_hash: str | None = None)

One canonical trace event. Construct via TraceEvent.new(...) so ids and timestamps are filled in; seal the hash chain with .chain(prev_hash).

Methods:

  • chain

    Link this event to its predecessor and seal its own digest (§10.2).

  • to_json

    Deterministic one-line JSON for JSONL storage (sorted keys, no spaces).

chain ¤
chain(prev_event_hash: str | None) -> TraceEvent

Link this event to its predecessor and seal its own digest (§10.2).

Source code in jdsl/trace/events.py
135
136
137
138
139
def chain(self, prev_event_hash: str | None) -> TraceEvent:
    """Link this event to its predecessor and seal its own digest (§10.2)."""
    self.prev_event_hash = prev_event_hash
    self.event_hash = self.compute_hash()
    return self
to_json ¤
to_json() -> str

Deterministic one-line JSON for JSONL storage (sorted keys, no spaces).

Source code in jdsl/trace/events.py
151
152
153
def to_json(self) -> str:
    """Deterministic one-line JSON for JSONL storage (sorted keys, no spaces)."""
    return json.dumps(self.to_dict(), sort_keys=True, separators=(",", ":"), default=str)

Trace Sinks¤

sink ¤

Trace sinks (design §35 PR1, §53 principle 2).

A TraceSink is where the runtime and adapters send TraceEvents. The default is NullTraceSink — the interpreter behaves exactly as before when nobody is capturing. Real sinks assign the per-episode sequence, seal the hash chain, and persist or buffer.

The telemetry-plane rule from §7.2 lives here: emit must be cheap and must fail open for pure observation. A sink that raises will never abort a run; SafeSink wraps any sink to guarantee that.

Classes:

  • TraceSink

    Where trace events go. emit returns the (possibly enriched) event so a

  • NullTraceSink

    Drops everything. The default in RunContext; keeps existing behavior

  • ListTraceSink

    Collects events in memory. Ideal for tests and for a single in-process run

  • SafeSink

    Wrap any sink so emit never raises (fail-open observation, §7.2). A

  • FanoutSink

    Emit to several sinks (e.g. an in-memory list plus a JSONL file). The first

TraceSink ¤

Bases: Protocol

Where trace events go. emit returns the (possibly enriched) event so a caller can read back the assigned sequence / hash for parent linkage.

NullTraceSink ¤

Drops everything. The default in RunContext; keeps existing behavior identical when capture is off.

ListTraceSink ¤

ListTraceSink()

Collects events in memory. Ideal for tests and for a single in-process run whose events are handed straight to the compiler.

Source code in jdsl/trace/sink.py
62
63
64
def __init__(self) -> None:
    self.events: list[TraceEvent] = []
    self._chain = _Chainer()

SafeSink ¤

SafeSink(inner: TraceSink, *, warn: bool = True)

Wrap any sink so emit never raises (fail-open observation, §7.2). A broken telemetry path must not take the agent down with it.

Source code in jdsl/trace/sink.py
79
80
81
82
def __init__(self, inner: TraceSink, *, warn: bool = True) -> None:
    self.inner = inner
    self.warn = warn
    self.errors = 0

FanoutSink ¤

FanoutSink(*sinks: TraceSink)

Emit to several sinks (e.g. an in-memory list plus a JSONL file). The first sink assigns sequence/hash; the rest see the already-stamped event.

Source code in jdsl/trace/sink.py
98
99
def __init__(self, *sinks: TraceSink) -> None:
    self.sinks = list(sinks)

JSONL and Replay¤

jsonl ¤

Append-only JSONL event storage (design §30 "JSONL for append-only event streams", §11.3 append-only Timeline).

One event per line, deterministic serialization (sorted keys). The Timeline is immutable: the compiler may reinterpret it but never rewrites it. JsonlTraceSink is a TraceSink that stamps the hash chain and appends; read_events streams a file back into TraceEvents.

Classes:

  • JsonlTraceSink

    Append events to a .jsonl file, one per line. Opens in append mode so an

Functions:

  • read_events

    Load a JSONL trace file into a list of TraceEvents (order preserved).

  • iter_events
  • verify_chain

    Verify per-episode hash chains (§10.2). Returns a list of human-readable

JsonlTraceSink ¤

JsonlTraceSink(path: str | Path, *, chain: _Chainer | None = None)

Append events to a .jsonl file, one per line. Opens in append mode so an interrupted run keeps whatever it already wrote (append-only, §11.3).

Source code in jdsl/trace/jsonl.py
24
25
26
27
28
def __init__(self, path: str | Path, *, chain: _Chainer | None = None) -> None:
    self.path = Path(path)
    self.path.parent.mkdir(parents=True, exist_ok=True)
    self._chain = chain or _Chainer()
    self._lock = threading.Lock()

read_events ¤

read_events(path: str | Path) -> list[TraceEvent]

Load a JSONL trace file into a list of TraceEvents (order preserved).

Source code in jdsl/trace/jsonl.py
38
39
40
def read_events(path: str | Path) -> list[TraceEvent]:
    """Load a JSONL trace file into a list of `TraceEvent`s (order preserved)."""
    return list(iter_events(path))

iter_events ¤

iter_events(path: str | Path) -> Iterator[TraceEvent]
Source code in jdsl/trace/jsonl.py
43
44
45
46
47
48
49
50
51
def iter_events(path: str | Path) -> Iterator[TraceEvent]:
    import json
    p = Path(path)
    with p.open("r", encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            yield TraceEvent.from_dict(json.loads(line))

verify_chain ¤

verify_chain(events: list[TraceEvent]) -> list[str]

Verify per-episode hash chains (§10.2). Returns a list of human-readable problems; an empty list means every chain is intact.

Source code in jdsl/trace/jsonl.py
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def verify_chain(events: list[TraceEvent]) -> list[str]:
    """Verify per-episode hash chains (§10.2). Returns a list of human-readable
    problems; an empty list means every chain is intact."""
    problems: list[str] = []
    tails: dict[str, str | None] = {}
    for e in events:
        if not e.verify_hash():
            problems.append(f"event {e.event_id} ({e.kind}) has a bad self-hash")
        expected_prev = tails.get(e.episode_id)
        if e.prev_event_hash != expected_prev:
            problems.append(
                f"event {e.event_id} in {e.episode_id} breaks the chain: "
                f"prev={e.prev_event_hash} expected={expected_prev}"
            )
        tails[e.episode_id] = e.event_hash
    return problems

replay ¤

Timeline reconstruction from a canonical event stream (design §11.3, §32.2).

The trace layer stores what happened. This module rebuilds usable views over an episode without interpreting it: the ordered timeline, the reconstructed blackboard state after each write, and the tool-call sequence. The compiler's verifier (§32.2) replays deterministic behavior against these views; here we only reconstruct, we never judge.

Classes:

  • ToolInvocation

    A single tool call reconstructed from started/completed/failed events.

  • Episode

    One reconstructed episode: its ordered events plus derived views.

Functions:

  • segment_episodes

    Group a flat event stream into episodes (§24 "segment episodes").

ToolInvocation dataclass ¤

ToolInvocation(logical_id: str | None, host_name: str | None, arguments: dict[str, Any], result: Any = None, error: Any = None, ok: bool = True, sequence: int = 0)

A single tool call reconstructed from started/completed/failed events.

Episode dataclass ¤

Episode(episode_id: str, events: list[TraceEvent] = list())

One reconstructed episode: its ordered events plus derived views.

Methods:

  • blackboard_states

    Cumulative blackboard snapshots after each blackboard.write, keyed by

  • tool_calls

    Pair up tool.call.started with the following completed/failed event.

blackboard_states ¤
blackboard_states() -> list[tuple[int, dict[str, Any]]]

Cumulative blackboard snapshots after each blackboard.write, keyed by the event sequence. Reconstructs state before each decision.

Source code in jdsl/trace/replay.py
71
72
73
74
75
76
77
78
79
80
def blackboard_states(self) -> list[tuple[int, dict[str, Any]]]:
    """Cumulative blackboard snapshots after each blackboard.write, keyed by
    the event sequence. Reconstructs `state before each decision`."""
    state: dict[str, Any] = {}
    out: list[tuple[int, dict[str, Any]]] = []
    for e in self.ordered():
        if e.kind == EventKind.BLACKBOARD_WRITE:
            state[e.payload["key"]] = e.payload.get("value")
            out.append((e.sequence, dict(state)))
    return out
tool_calls ¤
tool_calls() -> list[ToolInvocation]

Pair up tool.call.started with the following completed/failed event.

Source code in jdsl/trace/replay.py
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def tool_calls(self) -> list[ToolInvocation]:
    """Pair up tool.call.started with the following completed/failed event."""
    calls: list[ToolInvocation] = []
    pending: dict[str | None, ToolInvocation] = {}
    for e in self.ordered():
        if e.kind == EventKind.TOOL_CALL_STARTED:
            tool = e.payload.get("tool", {})
            inv = ToolInvocation(
                logical_id=tool.get("logical_id"),
                host_name=tool.get("host_name"),
                arguments=e.payload.get("arguments", {}) or {},
                sequence=e.sequence,
            )
            pending[e.event_id] = inv
            calls.append(inv)
        elif e.kind in (EventKind.TOOL_CALL_COMPLETED, EventKind.TOOL_CALL_FAILED):
            inv = pending.get(e.parent_event_id or "")
            if inv is None and calls:
                inv = calls[-1]  # best-effort pairing when parent linkage is absent
            if inv is not None:
                if e.kind == EventKind.TOOL_CALL_FAILED:
                    inv.ok = False
                    inv.error = e.payload.get("error")
                else:
                    inv.result = e.payload.get("result")
    return calls

segment_episodes ¤

segment_episodes(events: Iterable[TraceEvent]) -> list[Episode]

Group a flat event stream into episodes (§24 "segment episodes").

Source code in jdsl/trace/replay.py
103
104
105
106
107
108
109
110
111
112
113
114
def segment_episodes(events: Iterable[TraceEvent]) -> list[Episode]:
    """Group a flat event stream into episodes (§24 "segment episodes")."""
    order: list[str] = []
    by_id: dict[str, Episode] = {}
    for e in events:
        ep = by_id.get(e.episode_id)
        if ep is None:
            ep = Episode(episode_id=e.episode_id)
            by_id[e.episode_id] = ep
            order.append(e.episode_id)
        ep.events.append(e)
    return [by_id[i] for i in order]

Behavior IR¤

schema ¤

The serializable Behavior IR (design §21) and first-class residual Signature (§18/§19).

Humans keep authoring jdsl with the Python combinators; the compiler emits and a package ships this JSON IR. It is a distribution/implementation format, not the primary human DSL (§21). The IR is a restricted node vocabulary over the runtime's primitives — no embedded code, guards are the safe expression language (expr.py), and model leaves reference typed Signatures by id.

Classes:

SignatureInput dataclass ¤

SignatureInput(source: str, schema: dict[str, Any] = (lambda: {'type': 'string'})())

One input to a residual model leaf: where it comes from and its type.

SignatureOutput dataclass ¤

SignatureOutput(name: str, schema: dict[str, Any] = (lambda: {'type': 'string'})())

Signature dataclass ¤

Signature(id: str, kind: str = 'predict', inputs: dict[str, SignatureInput] = dict(), output: SignatureOutput | None = None, instruction: str = '', examples: list[dict[str, Any]] = list(), tools: list[str] = list(), context_policy: dict[str, Any] = dict(), validator: dict[str, Any] = (lambda: {'type': 'json_schema'})())

A typed interface between one tree leaf and the small model (§18.1). The string form predict("a -> b") remains the authoring shorthand (§19); this is the structured form packages carry.

IRNode dataclass ¤

IRNode(type: str, id: str | None = None)

Base IR node. type is the discriminator; id is the stable identity.

IRSequence dataclass ¤

IRSequence(type: str, id: str | None = None, children_: list[IRNode] = list())

Bases: IRComposite

IRSelector dataclass ¤

IRSelector(type: str, id: str | None = None, children_: list[IRNode] = list())

Bases: IRComposite

IROptional dataclass ¤

IROptional(type: str, id: str | None = None, child: IRNode | None = None)

Bases: IRDecorator

IRInvert dataclass ¤

IRInvert(type: str, id: str | None = None, child: IRNode | None = None)

Bases: IRDecorator

IRRepeat dataclass ¤

IRRepeat(type: str, id: str | None = None, child: IRNode | None = None, until: IRNode | None = None, max: int = 3)

Bases: IRDecorator

IRAction dataclass ¤

IRAction(type: str, id: str | None = None, tool: str = '', arguments: dict[str, Any] = dict(), store: str | None = None)

Bases: IRNode

IRGuard dataclass ¤

IRGuard(type: str, id: str | None = None, expression: dict[str, Any] = dict())

Bases: IRNode

IRGuardCall dataclass ¤

IRGuardCall(type: str, id: str | None = None, predicate: str = '', arguments: dict[str, Any] = dict())

Bases: IRNode

IRPredict dataclass ¤

IRPredict(type: str, id: str | None = None, signature: str = '')

Bases: IRNode

IRReact dataclass ¤

IRReact(type: str, id: str | None = None, signature: str = '')

Bases: IRNode

BehaviorIR dataclass ¤

BehaviorIR(root: IRNode, signatures: dict[str, Signature] = dict(), format: str = BEHAVIOR_FORMAT)

A whole compiled behavior tree plus the signatures its leaves reference.

expr ¤

The restricted guard-expression language (design §21.2).

Compiled packages must not ship arbitrary Python (§22.3). Guards are instead a small, safe, JSON expression tree evaluated against the blackboard. Operators are a fixed set; operands are literals, refs, or simple JSON paths. Anything a package can express here is reviewable and sandbox-safe by construction.

Expression grammar (all nodes are JSON objects with a single operator key)::

{"exists": <operand>}
{"eq": [<operand>, <operand>]}      # also neq, lt, lte, gt, gte
{"in": [<operand>, <operand>]}      # membership: left in right
{"and": [<expr>, ...]}              # also or
{"not": <expr>}

An is either: {"ref": "customer.id"} # resolved from the blackboard by path {"const": } # a literal value # shorthand for {"const": ...}

Paths support dotted keys and bracket indexing, and a $name index is itself a blackboard lookup (so orders[$selected_index].id works, §21.1)::

order.status
orders[0].id
orders[$selected_index].id

Classes:

  • ExprError

    A malformed guard expression (raised at validate/lower time, not runtime).

Functions:

  • resolve_path

    Resolve a dotted/bracketed path against the blackboard. Returns _MISSING

  • evaluate

    Evaluate a guard expression to a bool against the blackboard.

  • validate_expr

    Static-check an expression tree without a blackboard (§32.1). Returns a list

ExprError ¤

Bases: ValueError

A malformed guard expression (raised at validate/lower time, not runtime).

resolve_path ¤

resolve_path(path: str, blackboard: dict[str, Any]) -> Any

Resolve a dotted/bracketed path against the blackboard. Returns _MISSING (a sentinel) if any step is absent, so exists can distinguish absent from a stored None.

Source code in jdsl/ir/expr.py
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def resolve_path(path: str, blackboard: dict[str, Any]) -> Any:
    """Resolve a dotted/bracketed path against the blackboard. Returns `_MISSING`
    (a sentinel) if any step is absent, so `exists` can distinguish absent from a
    stored `None`."""
    if path in blackboard:
        return blackboard[path]
    cur: Any = _MISSING
    first = True
    for tok in _PATH_TOKEN.findall(path):
        if tok.startswith("["):
            index_expr = tok[1:-1].strip()
            index = _resolve_index(index_expr, blackboard)
            cur = _index_into(cur, index)
        elif first:
            cur = blackboard.get(tok, _MISSING)
        else:
            cur = _index_into(cur, tok)
        first = False
        if cur is _MISSING:
            return _MISSING
    return cur

evaluate ¤

evaluate(expr: Any, blackboard: dict[str, Any]) -> bool

Evaluate a guard expression to a bool against the blackboard.

Source code in jdsl/ir/expr.py
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def evaluate(expr: Any, blackboard: dict[str, Any]) -> bool:
    """Evaluate a guard expression to a bool against the blackboard."""
    if not isinstance(expr, dict) or len(expr) != 1:
        raise ExprError(f"expression must be a single-operator object, got {expr!r}")
    op, arg = next(iter(expr.items()))
    if op == "exists":
        return resolve_path(_ref_of(arg), blackboard) is not _MISSING
    if op == "not":
        return not evaluate(arg, blackboard)
    if op in ("and", "or"):
        if not isinstance(arg, list):
            raise ExprError(f"{op!r} needs a list of expressions")
        results = (evaluate(e, blackboard) for e in arg)
        return all(results) if op == "and" else any(results)
    if op == "in":
        left, right = _pair(op, arg)
        container = _operand(right, blackboard)
        try:
            return _operand(left, blackboard) in container
        except TypeError:
            return False
    if op in _BINARY:
        left, right = _pair(op, arg)
        a, b = _operand(left, blackboard), _operand(right, blackboard)
        try:
            return bool(_BINARY[op](a, b))
        except TypeError:
            return False
    raise ExprError(f"unknown operator {op!r}; allowed: {sorted(VALID_OPERATORS)}")

validate_expr ¤

validate_expr(expr: Any) -> list[str]

Static-check an expression tree without a blackboard (§32.1). Returns a list of problems; empty means well-formed.

Source code in jdsl/ir/expr.py
159
160
161
162
163
164
def validate_expr(expr: Any) -> list[str]:
    """Static-check an expression tree without a blackboard (§32.1). Returns a list
    of problems; empty means well-formed."""
    problems: list[str] = []
    _validate(expr, problems)
    return problems

lower ¤

Lower the Behavior IR into the existing runtime Node classes (design §35 PR11 "Lower IR into existing Node classes").

The IR is model-neutral and refers to logical capabilities, predicate ids, and signature ids. Lowering binds those to concrete runtime objects supplied by the host (§12.1 runtime binding): a missing capability makes lowering fail before any execution, never silently at run time.

Classes:

  • BindingError

    A required capability / predicate / signature was not supplied at load.

  • RuntimeBindings

    What the host provides to run a package (§12.1, §40). Tools are keyed by

Functions:

  • lower

    Lower a whole behavior IR to a runnable tree. Signatures come from

  • lower_node

BindingError ¤

Bases: ValueError

A required capability / predicate / signature was not supplied at load.

RuntimeBindings dataclass ¤

RuntimeBindings(tools: dict[str, Any] = dict(), predicates: dict[str, Any] = dict(), signatures: dict[str, Signature] = dict())

What the host provides to run a package (§12.1, §40). Tools are keyed by logical capability id; predicates by guard-call id; signatures by id.

lower ¤

lower(ir: BehaviorIR, bindings: RuntimeBindings) -> Node

Lower a whole behavior IR to a runnable tree. Signatures come from bindings if present, else from the IR's own signature table.

Source code in jdsl/ir/lower.py
75
76
77
78
79
80
def lower(ir: BehaviorIR, bindings: RuntimeBindings) -> Node:
    """Lower a whole behavior IR to a runnable tree. Signatures come from
    `bindings` if present, else from the IR's own signature table."""
    merged = RuntimeBindings(tools=bindings.tools, predicates=bindings.predicates,
                             signatures={**ir.signatures, **bindings.signatures})
    return lower_node(ir.root, merged)

lower_node ¤

lower_node(node: IRNode, b: RuntimeBindings) -> Node
Source code in jdsl/ir/lower.py
83
84
85
86
87
def lower_node(node: IRNode, b: RuntimeBindings) -> Node:
    out = _lower_dispatch(node, b)
    if node.id is not None:
        out.node_id = node.id
    return out

Packages¤

manifest ¤

Behavior-package metadata: manifest, tool contracts, and provenance (design §12, §22.2, §23).

A .jdsl is executable policy, so its metadata is first-class: the manifest declares required capabilities and verification status, tool contracts give each capability a portable logical identity and effect flags, and provenance answers "why does this node exist?" for audit.

Classes:

  • ToolEffects

    Effect flags for a capability (§12). Third-party MCP annotations are hints;

  • ToolContract

    A portable capability contract (§12): logical id + schemas + effects. Host

  • NodeProvenance

    Why a compiled node exists (§23) — traceable back to source evidence.

  • Manifest

    Package manifest (§22.2).

ToolEffects dataclass ¤

ToolEffects(read_only: bool = True, destructive: bool = False, idempotent: bool = True)

Effect flags for a capability (§12). Third-party MCP annotations are hints; critical properties should be contract-backed (§12, §15 E3).

ToolContract dataclass ¤

ToolContract(logical_id: str, input_schema: dict[str, Any] = (lambda: {'type': 'object'})(), output_schema: dict[str, Any] = (lambda: {'type': 'object'})(), effects: ToolEffects = ToolEffects(), description: str = '')

A portable capability contract (§12): logical id + schemas + effects. Host tool names (mcp__retail__get_order) bind to this logical id at runtime.

NodeProvenance dataclass ¤

NodeProvenance(node_id: str, behavior_candidate: str | None = None, evidence_grade: str = 'E0', source_episodes: list[str] = list(), contract_sources: list[str] = list(), compiler_model: str | None = None, rationale_summary: str = '')

Why a compiled node exists (§23) — traceable back to source evidence.

Manifest dataclass ¤

Manifest(name: str, version: str = '0.1.0', task_family: str = '', required_capabilities: list[str] = list(), runtime: dict[str, str] = (lambda: {'jdsl': '>=0.3'})(), source: dict[str, Any] = dict(), verification: dict[str, Any] = (lambda: {'status': 'unverified'})(), files: dict[str, str] = dict(), format: str = PACKAGE_FORMAT)

Package manifest (§22.2).

export ¤

The in-memory behavior package and its export to disk (design §22, §35 PR13).

The canonical development representation is an unpacked directory (§22.1); the transport form is a deterministic .jdsl ZIP (§22) whose bytes are a pure function of its contents (fixed timestamps, sorted entries) so the same package always hashes the same. No arbitrary code ships (§22.3): only restricted IR, signatures, expressions, contracts, tests, and provenance.

Classes:

  • BehaviorPackage

    A whole compiled behavior package held in memory (§22.1).

Functions:

  • export_dir

    Write the package as an unpacked directory (§22.1).

  • export_jdsl

    Write the package as a deterministic .jdsl ZIP (§22).

  • package_digest

    A stable digest over the deterministic ZIP bytes (for signing later, §22.4).

BehaviorPackage dataclass ¤

BehaviorPackage(manifest: Manifest, ir: BehaviorIR, tools: list[ToolContract] = list(), provenance: list[NodeProvenance] = list(), invariants: list[dict[str, Any]] = list(), postconditions: list[dict[str, Any]] = list(), tests: dict[str, list[dict[str, Any]]] = dict(), evidence_summary: dict[str, Any] = dict(), readme: str = '')

A whole compiled behavior package held in memory (§22.1).

Methods:

  • files

    The full set of package files as {relpath: text}, deterministically

files ¤
files() -> dict[str, str]

The full set of package files as {relpath: text}, deterministically serialized. This is what both the directory and the ZIP write.

Source code in jdsl/package/export.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def files(self) -> dict[str, str]:
    """The full set of package files as {relpath: text}, deterministically
    serialized. This is what both the directory and the ZIP write."""
    out: dict[str, str] = {}
    out["behavior.json"] = _dump(self.ir.to_dict())
    out["tools.json"] = _dump({"tools": [t.to_dict() for t in self.tools]})
    out["provenance.json"] = _dump({"nodes": [p.to_dict() for p in self.provenance]})
    for sid, sig in sorted(self.ir.signatures.items()):
        out[f"signatures/{sid}.json"] = _dump(sig.to_dict())
    out["contracts/invariants.json"] = _dump({"invariants": self.invariants})
    out["contracts/postconditions.json"] = _dump({"postconditions": self.postconditions})
    for name in ("replay", "guards", "signatures"):
        rows = self.tests.get(name, [])
        out[f"tests/{name}.jsonl"] = "".join(_dump(r, indent=None) + "\n" for r in rows)
    out["evidence/summary.json"] = _dump(self.evidence_summary)
    out["README.md"] = self.readme or _default_readme(self)
    # manifest last: it carries the digests of the other files (§22.2)
    self.manifest.files = {name: _sha256(text) for name, text in sorted(out.items())}
    out["manifest.json"] = _dump(self.manifest.to_dict())
    return out

export_dir ¤

export_dir(pkg: BehaviorPackage, path: str | Path) -> Path

Write the package as an unpacked directory (§22.1).

Source code in jdsl/package/export.py
61
62
63
64
65
66
67
68
69
def export_dir(pkg: BehaviorPackage, path: str | Path) -> Path:
    """Write the package as an unpacked directory (§22.1)."""
    root = Path(path)
    root.mkdir(parents=True, exist_ok=True)
    for rel, text in pkg.files().items():
        fp = root / rel
        fp.parent.mkdir(parents=True, exist_ok=True)
        fp.write_text(text, encoding="utf-8")
    return root

export_jdsl ¤

export_jdsl(pkg: BehaviorPackage, path: str | Path) -> Path

Write the package as a deterministic .jdsl ZIP (§22).

Source code in jdsl/package/export.py
72
73
74
75
76
77
78
79
def export_jdsl(pkg: BehaviorPackage, path: str | Path) -> Path:
    """Write the package as a deterministic `.jdsl` ZIP (§22)."""
    out = Path(path)
    if out.suffix != ".jdsl":
        out = out.with_suffix(".jdsl")
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_bytes(_zip_bytes(pkg.files()))
    return out

package_digest ¤

package_digest(pkg: BehaviorPackage) -> str

A stable digest over the deterministic ZIP bytes (for signing later, §22.4).

Source code in jdsl/package/export.py
82
83
84
def package_digest(pkg: BehaviorPackage) -> str:
    """A stable digest over the deterministic ZIP bytes (for signing later, §22.4)."""
    return "sha256:" + hashlib.sha256(_zip_bytes(pkg.files())).hexdigest()

load ¤

Load and bind a behavior package (design §40 runtime, §45 security model).

Loading treats a .jdsl package like software: verify the manifest format, verify file digests, structurally validate the IR (§32.1), and reject anything malformed before execution. Binding then attaches host-supplied tools and predicates (§12.1); a missing required capability fails the bind, never a run.

Classes:

  • PackageError

    A package failed to load: bad format, digest mismatch, or invalid IR.

  • LoadedPackage

    A verified, in-memory package ready to bind and run.

Functions:

  • load_package

    Load a package directory or .jdsl file and verify it structurally.

PackageError ¤

Bases: ValueError

A package failed to load: bad format, digest mismatch, or invalid IR.

LoadedPackage dataclass ¤

LoadedPackage(manifest: Manifest, ir: BehaviorIR, tools: list[ToolContract], provenance: list[NodeProvenance], root_dir: Path | None = None)

A verified, in-memory package ready to bind and run.

Methods:

  • as_root

    Wrap the bound tree in a Root so it runs like any authored skill (§40).

  • bind

    Bind capabilities/predicates and lower to a runnable tree (§12.1).

  • permissions

    The reads/writes a host should display before binding (§45).

as_root ¤
as_root(tools: dict[str, Any], predicates: dict[str, Any] | None = None, *, model_id: str | None = None) -> Root

Wrap the bound tree in a Root so it runs like any authored skill (§40).

Source code in jdsl/package/load.py
61
62
63
64
65
def as_root(self, tools: dict[str, Any], predicates: dict[str, Any] | None = None,
            *, model_id: str | None = None) -> Root:
    """Wrap the bound tree in a Root so it runs like any authored skill (§40)."""
    root = Root(name=self.manifest.name, child=self.bind(tools, predicates), model_id=model_id)
    return root
bind ¤
bind(tools: dict[str, Any], predicates: dict[str, Any] | None = None) -> Node

Bind capabilities/predicates and lower to a runnable tree (§12.1).

Source code in jdsl/package/load.py
53
54
55
56
57
58
59
def bind(self, tools: dict[str, Any], predicates: dict[str, Any] | None = None) -> Node:
    """Bind capabilities/predicates and lower to a runnable tree (§12.1)."""
    missing = [c for c in self.manifest.required_capabilities if c not in tools]
    if missing:
        raise PackageError(f"required capabilities not bound: {missing}")
    bindings = RuntimeBindings(tools=tools, predicates=predicates or {}, signatures=self.ir.signatures)
    return lower(self.ir, bindings)
permissions ¤
permissions() -> dict[str, list[str]]

The reads/writes a host should display before binding (§45).

Source code in jdsl/package/load.py
47
48
49
50
51
def permissions(self) -> dict[str, list[str]]:
    """The reads/writes a host should display before binding (§45)."""
    reads = [t.logical_id for t in self.tools if t.effects.read_only]
    writes = [t.logical_id for t in self.tools if not t.effects.read_only]
    return {"reads": sorted(reads), "writes": sorted(writes)}

load_package ¤

load_package(path: str | Path, *, verify_digests: bool = True) -> LoadedPackage

Load a package directory or .jdsl file and verify it structurally.

Source code in jdsl/package/load.py
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def load_package(path: str | Path, *, verify_digests: bool = True) -> LoadedPackage:
    """Load a package directory or `.jdsl` file and verify it structurally."""
    p = Path(path)
    files = _read_zip(p) if p.is_file() else _read_dir(p)

    if "manifest.json" not in files:
        raise PackageError("package has no manifest.json")
    manifest = Manifest.from_dict(json.loads(files["manifest.json"]))
    if manifest.format not in _SUPPORTED_FORMATS:
        raise PackageError(f"unsupported package format {manifest.format!r}")

    if verify_digests:
        _verify_digests(manifest, files)

    if "behavior.json" not in files:
        raise PackageError("package has no behavior.json")
    signatures = _load_signatures(files)
    ir = BehaviorIR.from_dict(json.loads(files["behavior.json"]), signatures)

    report = validate_ir(ir, required_capabilities=set(manifest.required_capabilities))
    if not report.ok:
        raise PackageError("invalid package IR:\n  " + "\n  ".join(report.problems))

    tools = [ToolContract.from_dict(t) for t in json.loads(files.get("tools.json", '{"tools":[]}'))["tools"]]
    prov = [NodeProvenance.from_dict(n)
            for n in json.loads(files.get("provenance.json", '{"nodes":[]}'))["nodes"]]
    return LoadedPackage(manifest=manifest, ir=ir, tools=tools, provenance=prov,
                         root_dir=p if p.is_dir() else None)