zettelkasten.synapse.memdsl.parser

zettelkasten.synapse.memdsl.parser

Deterministic, injection-safe (de)serializers for the MemDSL contract.

This module is the wire half of the contract: it turns the in-memory types from :mod:zettelkasten.synapse.memdsl.schema into a compact, line-oriented text form and back, in all three directions:

• RENDER (engine → agent): :func:serialize_envelope / :func:parse_envelope.
• OUTPUT (agent → engine): :func:serialize_output / :func:parse_output.
• ACTION-RESPONSE (engine → agent, mid-loop): :func:serialize_response / :func:parse_response.

Two properties are load-bearing:

• Deterministic — every serializer is a pure function of its input, emitting fields in a fixed order, so identical input yields byte-identical output and parse(serialize(x)) == x.
• Injection-safe — untrusted node bodies are framed line-by-line with a | continuation prefix, so a body line that looks like a control token (@answer, ground(n1), node …) can NEVER be read as structure. The three directions also carry distinct headers, so a render envelope fed to :func:parse_output is rejected outright rather than mined for actions. This mirrors the untrusted-content framing used in :mod:zettelkasten.synapse.synthesis, moved from a prompt convention into the grammar itself.

The grammar is a small hand-written line format rather than a parser-generator dependency: robust, fully tested, and easy to audit for the injection property.

serialize_envelope

serialize_envelope(env: Envelope) -> str

Serialize a render :class:Envelope to MemDSL text (deterministic).

Source code in zettelkasten/synapse/memdsl/parser.py

def serialize_envelope(env: Envelope) -> str:
    """Serialize a render :class:`Envelope` to MemDSL text (deterministic)."""
    lines = [RENDER_HEADER]
    if env.focus is not None:
        lines.append(_line(_lit("focus"), _emit(env.focus, _FOCUS)))
    for node in env.nodes:
        lines.extend(_serialize_node(node))
    for gap in env.gaps:
        lines.append(_serialize_gap(gap))
    return "\n".join(lines) + "\n"

parse_envelope

parse_envelope(text: str) -> Envelope

Parse MemDSL render text back into an :class:Envelope.

Raises :class:MemDSLParseError on a bad/foreign header or malformed block.

Source code in zettelkasten/synapse/memdsl/parser.py

def parse_envelope(text: str) -> Envelope:
    """Parse MemDSL render text back into an :class:`Envelope`.

    Raises :class:`MemDSLParseError` on a bad/foreign header or malformed block.
    """
    lines = _split_lines(text)
    if not lines:
        raise MemDSLParseError("empty envelope")
    version = _parse_header(lines[0], RENDER_HEADER)
    nodes: list[RenderedNode] = []
    gaps: list[GapToken] = []
    focus: str | None = None
    i = 1
    while i < len(lines):
        line = lines[i]
        if line.startswith("focus "):
            focus = line[len("focus ") :].strip()
            i += 1
        elif line.startswith("node "):
            node, i = _parse_node_block(lines, i)
            nodes.append(node)
        elif line.startswith("gap "):
            gaps.append(_parse_gap(line))
            i += 1
        elif line == "":
            i += 1
        else:
            raise MemDSLParseError(f"unexpected line in envelope: {line!r}")
    return Envelope(nodes=nodes, gaps=gaps, focus=focus, format_version=version)

serialize_response

serialize_response(resp: ActionResponse) -> str

Serialize an :class:ActionResponse fragment to MemDSL text.

v1 emits status + read results (nodes/gaps); the reserved mut/ impact lines only appear when future write actions populate them.

Source code in zettelkasten/synapse/memdsl/parser.py

def serialize_response(resp: ActionResponse) -> str:
    """Serialize an :class:`ActionResponse` fragment to MemDSL text.

    v1 emits ``status`` + read results (nodes/gaps); the reserved ``mut``/
    ``impact`` lines only appear when future write actions populate them.
    """
    lines = [RESPONSE_HEADER, _line(_lit("status"), _emit(resp.status, _STATUS))]
    for node in resp.nodes:
        lines.extend(_serialize_node(node))
    for gap in resp.gaps:
        lines.append(_serialize_gap(gap))
    for mutation in resp.mutations:
        lines.append(_serialize_mutation(mutation))
    if resp.impact:
        lines.append(_line(_lit("impact"), _join(_lit(", "), [_emit(x, _IMPACT) for x in resp.impact])))
    return "\n".join(lines) + "\n"

parse_response

parse_response(text: str) -> ActionResponse

Parse an :class:ActionResponse fragment back from MemDSL text.

Source code in zettelkasten/synapse/memdsl/parser.py

def parse_response(text: str) -> ActionResponse:
    """Parse an :class:`ActionResponse` fragment back from MemDSL text."""
    lines = _split_lines(text)
    if not lines:
        raise MemDSLParseError("empty action response")
    version = _parse_header(lines[0], RESPONSE_HEADER)
    status = "ok"
    nodes: list[RenderedNode] = []
    gaps: list[GapToken] = []
    mutations: list[Mutation] = []
    impact: list[str] = []
    i = 1
    while i < len(lines):
        line = lines[i]
        if line.startswith("status "):
            status = line[len("status ") :].strip()
            i += 1
        elif line.startswith("node "):
            node, i = _parse_node_block(lines, i)
            nodes.append(node)
        elif line.startswith("gap "):
            gaps.append(_parse_gap(line))
            i += 1
        elif line.startswith("mut "):
            mutations.append(_parse_mutation(line))
            i += 1
        elif line.startswith("impact "):
            impact = [p.strip() for p in line[len("impact ") :].split(",") if p.strip()]
            i += 1
        elif line == "":
            i += 1
        else:
            raise MemDSLParseError(f"unexpected line in action response: {line!r}")
    return ActionResponse(
        status=status,
        nodes=nodes,
        gaps=gaps,
        mutations=mutations,
        impact=impact,
        format_version=version,
    )

serialize_output

serialize_output(out: AgentOutput) -> str

Serialize an :class:AgentOutput (actions + terminal decision) to MemDSL text.

Raises :class:MemDSLParseError if both a reasoning answer and an abstain are present (they are mutually exclusive terminal decisions).

The conclude line is emitted whenever the reasoning-DAG has claims: the conclusion is canonicalized to the last claim id at DAG construction (see :meth:ReasoningDAG.__post_init__), so there is no empty-conclusion case for the serializer to drop. This is what makes parse(serialize(x)) == x hold, closing the old asymmetry where the serializer omitted conclude but the parser back-filled it.

Source code in zettelkasten/synapse/memdsl/parser.py

def serialize_output(out: AgentOutput) -> str:
    """Serialize an :class:`AgentOutput` (actions + terminal decision) to MemDSL text.

    Raises :class:`MemDSLParseError` if both a reasoning ``answer`` and an
    ``abstain`` are present (they are mutually exclusive terminal decisions).

    The ``conclude`` line is emitted whenever the reasoning-DAG has claims: the
    conclusion is canonicalized to the last claim id at DAG construction (see
    :meth:`ReasoningDAG.__post_init__`), so there is no empty-conclusion case for
    the serializer to drop. This is what makes ``parse(serialize(x)) == x`` hold,
    closing the old asymmetry where the serializer omitted ``conclude`` but the
    parser back-filled it.
    """
    if out.reasoning is not None and out.abstain is not None:
        raise MemDSLParseError("an output cannot both @answer and @abstain")
    if out.format_version == LEGACY_OUTPUT_FORMAT_VERSION:
        header = OUTPUT_HEADER
        if out.reasoning is not None and (
            out.reasoning.prose or out.reasoning.prose_citations
        ):
            raise MemDSLParseError("OUTPUT v1 cannot serialize prose cite bindings")
    elif out.format_version == OUTPUT_FORMAT_VERSION:
        header = OUTPUT_HEADER_V2
        if out.reasoning is not None:
            out.reasoning.validate()
    else:
        raise MemDSLParseError(
            f"unsupported OUTPUT version {out.format_version!r}; expected "
            f"{LEGACY_OUTPUT_FORMAT_VERSION!r} or {OUTPUT_FORMAT_VERSION!r}"
        )

    lines = [header]
    for action in out.actions:
        lines.append(_serialize_action(action))
    if out.reasoning is not None:
        lines.append("@answer")
        for claim in out.reasoning.claims:
            lines.append(_serialize_claim(claim))
        if out.format_version == OUTPUT_FORMAT_VERSION:
            lines.append(_line(_lit("prose"), _lit("::"), _emit(out.reasoning.prose, _PROSE_TEXT)))
            for citation in sorted(
                out.reasoning.prose_citations, key=lambda item: item.span
            ):
                start, end = citation.span
                quote = out.reasoning.prose[start:end]
                lines.append(
                    _line(
                        _lit("cite"),
                        _emit(citation.claim_ids[0], _CLAIM_ID),
                        _lit("::"),
                        _emit(quote, _PROSE_QUOTE),
                    )
                )
        if out.reasoning.conclusion:
            # ``conclude`` is ``[len:].strip()``-parsed and names a claim id (a
            # whitespace-free token), so guard it like a claim id.
            lines.append(_line(_lit("conclude"), _emit(out.reasoning.conclusion, _CONCLUDE_ID)))
    if out.abstain is not None:
        # The reason is the anchored head of the ``@abstain`` grammar
        # (:data:`_ABSTAIN_RE`): it must not carry the `` :: `` reason/detail
        # separator or a line break (either would let the parser rewrite the
        # reason/detail boundary or split a fresh structural line), and must be
        # non-empty. It also forbids leading/trailing whitespace: the regex
        # capture does NOT strip the reason (so such whitespace would technically
        # round-trip), but it is rejected as harmless over-strictness — a reason
        # label carries no surrounding whitespace. The detail, like every other
        # rest-of-line detail, is captured verbatim, so its whitespace round-trips.
        parts = [_lit("@abstain"), _emit(out.abstain.reason, _ABSTAIN_REASON)]
        if out.abstain.detail:
            parts.append(_lit("::"))
            parts.append(_emit(out.abstain.detail, _ABSTAIN_DETAIL))
        lines.append(_line(*parts))
    return "\n".join(lines) + "\n"

parse_output

parse_output(text: str) -> AgentOutput

Parse agent OUTPUT text into an :class:AgentOutput.

The required @memdsl.out/<ver> header is what makes this injection-safe: a render envelope or a raw untrusted node body has a different (or no) header, so it is rejected here rather than mined for actions. Within an @answer block, v1 carries claims/conclude and v2 additionally requires prose plus exact cite bindings.

A PURELY SYNTACTIC, MARKER-DELIMITED pre-parse recovery (:func:_recover_output_envelope) first unwraps a single, exactly-matched enclosing markdown code fence (with only blank lines outside it), so a production LLM's common near-miss still yields an answer. It trims NOTHING it would have to classify as prose-vs-structural — only a matched fence pair's wrapper and leading/trailing blank lines — and hands the envelope interior VERBATIM to the strict parse below, which requires the first non-blank line to be the header and consumes to the end or refuses. So recovery can launder nothing: a pristine structural line OUTSIDE the recovered core (pre-header preamble, or outside a fence) is never dropped — it stays in the interior and is rejected here, or lies outside the fence and refuses — and everything below fails-closed.

Exactly one terminal decision is allowed: @answer XOR @abstain, once. A second terminal, or any action AFTER a terminal, is a :class:MemDSLParseError — a turn ends at its single terminal decision, so a trailing action can never silently ride along after it.

Source code in zettelkasten/synapse/memdsl/parser.py

def parse_output(text: str) -> AgentOutput:
    """Parse agent OUTPUT text into an :class:`AgentOutput`.

    The required ``@memdsl.out/<ver>`` header is what makes this injection-safe:
    a render envelope or a raw untrusted node body has a different (or no) header,
    so it is rejected here rather than mined for actions. Within an ``@answer``
    block, v1 carries claims/``conclude`` and v2 additionally requires
    ``prose`` plus exact ``cite`` bindings.

    A PURELY SYNTACTIC, MARKER-DELIMITED pre-parse recovery
    (:func:`_recover_output_envelope`) first unwraps a single, exactly-matched
    enclosing markdown code fence (with only blank lines outside it), so a
    production LLM's common near-miss still yields an answer. It trims NOTHING it
    would have to classify as prose-vs-structural — only a matched fence pair's
    wrapper and leading/trailing blank lines — and hands the envelope interior
    VERBATIM to the strict parse below, which requires the first non-blank line to
    be the header and consumes to the end or refuses. So recovery can launder
    nothing: a pristine structural line OUTSIDE the recovered core (pre-header
    preamble, or outside a fence) is never dropped — it stays in the interior and
    is rejected here, or lies outside the fence and refuses — and everything below
    fails-closed.

    Exactly one terminal decision is allowed: ``@answer`` XOR ``@abstain``, once.
    A second terminal, or any action AFTER a terminal, is a
    :class:`MemDSLParseError` — a turn ends at its single terminal decision, so a
    trailing action can never silently ride along after it.
    """
    lines = _split_lines(_recover_output_envelope(text))
    if not lines:
        raise MemDSLParseError("empty output")
    header = lines[0].strip()
    if header == OUTPUT_HEADER_V2:
        version = OUTPUT_FORMAT_VERSION
    elif header == OUTPUT_HEADER_V1:
        version = LEGACY_OUTPUT_FORMAT_VERSION
    else:
        raise MemDSLParseError(
            f"expected MemDSL header {OUTPUT_HEADER_V2!r} or {OUTPUT_HEADER_V1!r}, got {header!r}"
        )
    actions: list[Action] = []
    reasoning: ReasoningDAG | None = None
    abstain: Abstention | None = None
    terminal_seen = False
    i = 1
    while i < len(lines):
        line = lines[i]
        if line == "":
            i += 1
            continue
        if line == "@answer" or line.startswith("@answer "):
            if terminal_seen:
                raise MemDSLParseError(
                    "an output has at most one terminal decision (@answer/@abstain)"
                )
            reasoning, i = _parse_answer_block(lines, i + 1, version=version)
            terminal_seen = True
        elif line.startswith("@abstain"):
            if terminal_seen:
                raise MemDSLParseError(
                    "an output has at most one terminal decision (@answer/@abstain)"
                )
            abstain = _parse_abstain(line)
            terminal_seen = True
            i += 1
        else:
            if terminal_seen:
                raise MemDSLParseError(
                    f"action after the terminal decision is not allowed: {line!r}"
                )
            actions.append(_parse_action(line))
            i += 1
    return AgentOutput(
        actions=actions,
        reasoning=reasoning,
        abstain=abstain,
        format_version=version,
    )

assign_refs

assign_refs(nodes: list[RenderedNode], *, prefix: str = 'n', start: int = 1) -> list[RenderedNode]

Assign stable, deterministic envelope-local refs (n1, n2, …) in order.

Mutates each node's ref in list order and returns the list, so the same node sequence always yields the same refs — the "stable id assignment" half of deterministic rendering. Ordering of the nodes themselves is the caller's responsibility (the serializer preserves it).

Source code in zettelkasten/synapse/memdsl/parser.py

def assign_refs(nodes: list[RenderedNode], *, prefix: str = "n", start: int = 1) -> list[RenderedNode]:
    """Assign stable, deterministic envelope-local refs (``n1``, ``n2``, …) in order.

    Mutates each node's ``ref`` in list order and returns the list, so the same
    node sequence always yields the same refs — the "stable id assignment" half
    of deterministic rendering. Ordering of the nodes themselves is the caller's
    responsibility (the serializer preserves it).
    """
    for offset, node in enumerate(nodes):
        node.ref = f"{prefix}{start + offset}"
    return nodes