Gate clients

doublegate_sdk connects to a gate only when you construct a transport and call a method. Nothing is discovered from the environment; no credential is read on import; nothing retries on its own. Two clients share one transport layer and one operation table (ADR-0074):

client	who	what it can do	what it can never do
KnowledgeClient	an agent, a connector, an application	remember, learn, propose a skill or a bundle, recall, read status/why/tip/relations/approval/comments, annotate, object, resolve, defer, rank or hide another writer's row	sign, promote, demote, reject, relate, ban, mint an operator proof
CurationClient	the operator (a human, or a script the human runs)	approve, hold, reject, veto, promote, relate/supersede, override the tip, rank, hide, ban, away/back, submit, semantic review, the organization's keys	act without the operator's signature

The gate is not callable by the thing being gated. The knowledge client's transport refuses a deciding verb before any I/O; the gate refuses it again from the identity it derived off the connection (I2). A successful proposal is a receipt, not an admission: every result carries proposals_are_admission = False.

These two clients speak the gate's dg.* verbs over its Unix socket or its POST /rpc door. The third door, POST /mcp, is the tool catalog an MCP host sees; doublegate_sdk.client (Gate client (HTTP/MCP)) speaks it with HttpMcpTransport, exposes only the tools the maintained client tier serves, and raises the same GateError.

Discover the API

python -m doublegate_sdk describe-client prints both clients' methods from their live signatures, the operation table (rpc, mutates, proof, role, scope), the three scopes, the error kinds and the two code tables. It describes this SDK. KnowledgeClient.describe() is what a running gate answers (dg.describe), parsed into Capabilities.

Transports and scopes

from doublegate_sdk.transport import UnixSocketTransport, HttpTransport

local = UnixSocketTransport('/explicit/gate/daemon.sock', timeout=5, scope='knowledge')
console = HttpTransport('http://127.0.0.1:8480', bearer=console_token, scope='knowledge')
org = HttpTransport('https://org.example', bearer=admin_key, scope='curation')

scope is the widest class of verb the transport will emit: read (the default; every write is writes_disabled), knowledge (an agent's verbs), or curation (the operator's). A verb outside the scope raises GateError('forbidden_operation') before a connection is opened; an unknown verb is a ValueError. Both transports share one deadline for connect, send and every read (10 s by default), a 1 MiB request cap and a 1 MiB response cap, duplicate-key rejection and the same code table. with_timeout(seconds) returns a copy for a short liveness probe. A caller-supplied GateTransport (call(method, params) -> dict) owns its own allowlist.

HttpTransport posts one JSON-RPC message to {base}/rpc with a bearer: the client gate's console token on its loopback listener, or an organization gate's admin key. A JSON-RPC refusal rides a 200; the door's own refusals are HTTP statuses (auth 401, scope 403, request_too_large 413, busy 429/503 with Retry-After kept as retry_after_ms). It is not an SSO adapter and it adds no authorization: the gate still checks every request.

Errors

GateError.kind is what you act on. code keeps the server's JSON-RPC number; retry_after_ms is set on busy; outcome_unknown is true when a write may have been retained although the reply was lost; detail carries the server's message, capped at 512 characters, and never appears in str(exc).

kind	when
unavailable, timeout, unsupported_transport	no gate answered in time
invalid_response, response_too_large, request_too_large	the frame, not the gate
writes_disabled, forbidden_operation, proof_required, page_limit	refused on this side, before I/O
identity, invalid_params, unsupported_operation, busy, banned, refused, remote_error	the gate's answer (-32000, -32600/-32602, -32601, -32006, -32009, -32010..-32014, other)
auth, scope	the HTTP /rpc door (401, 403)
unauthorized, forbidden, tool_error, redirect_refused	the HTTP /mcp door (mcp-client)

The same two tables live in the constellation app's rpc.ts; change both or neither.

The knowledge client

from doublegate_sdk.knowledge import KnowledgeClient

agent = KnowledgeClient(local)
if agent.present():
    caps = agent.describe()                       # Capabilities: verbs, params, content types
    memo = agent.remember('The lab labels basalt by collection date.',
                          content_type='memory', trust_class='T-4', source_uri='agent://field/42')
    fact = agent.learn('Basalt labels are durable.', source_uri='agent://field/42',
                       evidence=[memo.artifact_id], kind='derived_fact')
    agent.annotate(fact.artifact_id, 'derived from the 2026-09 survey', kind='note')
    row = agent.wait_for(fact.artifact_id, {'ACTIVE', 'REJECTED'}, timeout=30)
    hits = agent.recall('basalt', limit=5, include_provisional=True)

• remember proposes one artifact: bytes travel base64 and unchanged; the gate scans before it answers and holds the row (L1_SCANNED or L1B_FLAGGED). No writer identity is ever sent.
• learn is a belief distilled from evidence: kind is derived_fact, summary or memory; evidence becomes derives_from, which is provenance and never hides a parent; the default trust class is T-5, a proposal about the world. replaces=<artifact id> adds that claim to the provenance and leaves a note "proposes: supersedes …" on the new row. Only a gate decision replaces belief; supersedes itself is the operator's relation.
• propose_skill, propose_prompt_template and propose_bundle (a ZIP, inspected locally first, then dg.ingest_bundle beside its fetch record).
• recall always excludes the caller's own unreviewed rows; provisional, superseded and hidden rows only on explicit request, each hit keeping its flags. No score is relabelled as trust.
• annotate, raise_objection, resolve, retract_comment are remarks on the record (ADR-0072). Only a human's objection holds a promotion; an agent's is a remark the reviewer will see.
• rank and hide weight a row for an audience; the gate refuses them on the caller's own rows (I2).
• inventory / inventory_pages list retained metadata with scope and coverage, bounded by max_pages; presence never implies admission.

The curation client

from doublegate_sdk.curation import CurationClient
from doublegate_sdk.proof import ServerMinted, SignerProof

# on the client gate's host, the daemon mints the proof from the key it holds
operator = CurationClient(UnixSocketTransport(sock, scope='curation'), ServerMinted())
# anywhere else, the operator signs the nonce with a function the SDK never sees inside
operator = CurationClient(org, SignerProof(sign=my_ed25519_key.sign))

operator.approve(aid, note='matches the survey')        # dg.sign {promote: true}
operator.veto(other, 'contradicts the 2025 audit')      # dg.demote
operator.supersede(new=aid, old=other)                  # dg.relate supersedes
operator.reject(third, 'duplicate', in_favour_of=aid)

Every operator verb fetches a fresh dg.challenge nonce, has the provider sign it, and sends {nonce, sig} with the call; nonces are single-use, so approve is one proof for the signature and the promotion it was for. Without a provider the operator verbs raise proof_required before I/O. rank, hide and comment attach a proof when a provider is present. The SDK never loads keys/operator.ed25519 and never depends on a crypto library: the signer is yours. operator.knowledge reads the same gate the agent's way.

The organization gate's REST doors

doublegate_sdk.org_rest moves the submission body that doublegate_sdk.submission encodes: put_submission returns a SubmissionOutcome whose result is one of accepted (202/200), retry (503/429 with the door's delay, or an unexpected status), stopped (401/403: a key problem does not become correct by waiting) or refused (400/413/422: this body, not the key). pull_outcomes pages GET /outcomes by cursor. Policy — backoff, cursors, what to do about retry — stays with the caller.

Verified scope

Unit tests drive both transports against a real AF_UNIX listener and a loopback http.server: the allowlist before I/O, the shared deadline, byte caps, malformed and duplicate-key replies, lost-reply outcomes, the code tables, every client method against a recording transport, the proof flow for each operator verb. examples/verify_knowledge_lifecycle.py runs a real Client Gate in-process with the deterministic StubBackend: proposal → held → duplicate → admission → provisional recall → a learning with provenance. It is not live-model, cross-principal, Windows or organization-delivery evidence.

doublegate_sdk.knowledge

The knowledge client: what an agent may do with a gate (ADR-0074 d2, d4).

Propose, learn, ask, annotate, weight for an audience. Never decide: no method here signs, promotes, demotes, rejects or relates, and the transport refuses those verbs before I/O. Identity is the connection's — nothing here takes or sends a writer identity (I2). A successful proposal is a receipt, not an admission; proposals_are_admission is False on every result.

LEARNING_KINDS module-attribute

LEARNING_KINDS: frozenset[str] = frozenset({'derived_fact', 'summary', 'memory'})

COMMENT_KINDS module-attribute

COMMENT_KINDS: frozenset[str] = frozenset({'note', 'question', 'objection', 'resolution'})

Proposal dataclass

The gate's receipt for one proposal. state is the gate's word, not ours.

Source code in src/doublegate_sdk/knowledge.py

@dataclass(frozen=True)
class Proposal:
    """The gate's receipt for one proposal. ``state`` is the gate's word, not ours."""
    artifact_id: str
    state: str
    duplicate: bool = False
    findings_count: int | None = None
    raw: dict[str, Any] = field(default_factory=dict, repr=False)
    proposals_are_admission: ClassVar[bool] = False

    @classmethod
    def from_response(cls, result: Any) -> Proposal:
        if (not isinstance(result, dict) or not isinstance(result.get('artifact_id'), str) or not result['artifact_id']
                or not isinstance(result.get('state'), str) or not result['state']):
            raise GateError('invalid_response', outcome_unknown=True)
        fc = result.get('findings_count')
        return cls(result['artifact_id'], result['state'], bool(result.get('duplicate', False)),
                   fc if type(fc) is int else None, dict(result))

artifact_id instance-attribute

artifact_id: str

state instance-attribute

state: str

duplicate class-attribute instance-attribute

duplicate: bool = False

findings_count class-attribute instance-attribute

findings_count: int | None = None

raw class-attribute instance-attribute

raw: dict[str, Any] = field(default_factory=dict, repr=False)

proposals_are_admission class-attribute

proposals_are_admission: bool = False

from_response classmethod

from_response(result: Any) -> Proposal

Source code in src/doublegate_sdk/knowledge.py

@classmethod
def from_response(cls, result: Any) -> Proposal:
    if (not isinstance(result, dict) or not isinstance(result.get('artifact_id'), str) or not result['artifact_id']
            or not isinstance(result.get('state'), str) or not result['state']):
        raise GateError('invalid_response', outcome_unknown=True)
    fc = result.get('findings_count')
    return cls(result['artifact_id'], result['state'], bool(result.get('duplicate', False)),
               fc if type(fc) is int else None, dict(result))

Learning dataclass

Bases: Proposal

A learning: content plus the evidence it derives from. proposed_supersedes names the claim it would replace; only a gate decision replaces belief (ADR-0049 d5).

Source code in src/doublegate_sdk/knowledge.py

@dataclass(frozen=True)
class Learning(Proposal):
    """A learning: content plus the evidence it derives from. ``proposed_supersedes``
    names the claim it would replace; only a gate decision replaces belief (ADR-0049 d5)."""
    kind: str = 'derived_fact'
    derives_from: tuple[str, ...] = ()
    proposed_supersedes: str | None = None

kind class-attribute instance-attribute

kind: str = 'derived_fact'

derives_from class-attribute instance-attribute

derives_from: tuple[str, ...] = ()

proposed_supersedes class-attribute instance-attribute

proposed_supersedes: str | None = None

KnowledgeClient

An agent's view of a gate through an explicit transport of scope read or knowledge.

Source code in src/doublegate_sdk/knowledge.py

class KnowledgeClient:
    """An agent's view of a gate through an explicit transport of scope ``read`` or ``knowledge``."""

    def __init__(self, transport: GateTransport, *, probe: GateTransport | None = None):
        scope = getattr(transport, 'scope', None)
        if scope == CURATION:
            raise ValueError('the knowledge client refuses a curation-scoped transport: an agent never decides')
        self._transport = transport
        self._probe = probe if probe is not None else transport
        self._capabilities: Capabilities | None = None

    # ---- liveness and the catalog ----

    def present(self) -> bool:
        """Is a gate answering? Never raises; a refusal or an outage is ``False``."""
        try:
            answer = self._probe.call('dg.ping', {})
        except GateError:
            return False
        return isinstance(answer, dict) and bool(answer.get('pong'))

    def ping(self) -> dict[str, Any]:
        return self._object(self._transport.call('dg.ping', {}))

    def describe(self, *, refresh: bool = False) -> Capabilities:
        if self._capabilities is None or refresh:
            self._capabilities = Capabilities.from_describe(self._object(self._transport.call('dg.describe', {})))
        return self._capabilities

    # ---- reads ----

    def status(self, artifact_id: str | None = None) -> dict[str, Any]:
        params = {} if artifact_id is None else {'artifact_id': _text(artifact_id, 'artifact_id')}
        return self._object(self._transport.call('dg.status', params))

    def why(self, artifact_id: str) -> dict[str, Any]:
        return self._one('dg.why', artifact_id)

    def tip(self, artifact_id: str) -> dict[str, Any]:
        return self._one('dg.tip', artifact_id)

    def relations(self, artifact_id: str) -> dict[str, Any]:
        return self._one('dg.relations', artifact_id)

    def approval(self, artifact_id: str) -> dict[str, Any]:
        return self._one('dg.approval', artifact_id)

    def node(self, artifact_id: str) -> dict[str, Any]:
        return self._one('dg.node', artifact_id)

    def comments(self, artifact_id: str, *, include_retracted: bool = False) -> dict[str, Any]:
        if type(include_retracted) is not bool:
            raise ValueError('include_retracted must be boolean')
        return self._object(self._transport.call('dg.comments', {'artifact_id': _text(artifact_id, 'artifact_id'),
                                                                 'include_retracted': include_retracted}))

    def pending(self, *, limit: int = 100) -> dict[str, Any]:
        if not _integer(limit, 1):
            raise ValueError('limit must be a positive integer')
        return self._object(self._transport.call('dg.pending', {'limit': limit}))

    def inventory(self, *, limit: int = 100, offset: int = 0, **filters: Any) -> InventoryPage:
        if not _integer(limit, 1) or not _integer(offset, 0):
            raise ValueError('limit must be positive and offset nonnegative integers')
        if set(filters) - _FILTERS:
            raise ValueError('unsupported inventory parameter')
        result = self._transport.call('dg.inventory', {'limit': limit, 'offset': offset, **filters})
        return InventoryPage.from_response(self._object(result), requested_offset=offset)

    def inventory_pages(self, *, limit: int = 100, max_pages: int = 10, **filters: Any) -> Iterator[InventoryPage]:
        if not _integer(max_pages, 1):
            raise ValueError('max_pages must be a positive integer')
        offset = 0
        for _ in range(max_pages):
            page = self.inventory(limit=limit, offset=offset, **filters)
            yield page
            if not page.has_more:
                return
            if page.next_offset is None:
                raise GateError('invalid_response')
            offset = page.next_offset
        raise GateError('page_limit')

    def recall(self, query: str, *, limit: int = 10, spaces: list[str] | None = None,
               include_provisional: bool = False, include_superseded: bool = False,
               include_hidden: bool = False, audience: str | None = None,
               mode: str | None = None) -> dict[str, Any]:
        """Served knowledge. The caller's own unreviewed rows are always excluded;
        provisional, superseded and hidden rows only on explicit request, and
        each hit keeps its flags — no score is relabelled as trust."""
        if not isinstance(query, str) or not query.strip() or not _integer(limit, 1):
            raise ValueError('nonempty query and positive integer limit required')
        for name, flag in (('include_provisional', include_provisional), ('include_superseded', include_superseded),
                           ('include_hidden', include_hidden)):
            if type(flag) is not bool:
                raise ValueError(f'{name} must be boolean')
        if spaces is not None and (not isinstance(spaces, list) or not spaces
                                   or any(not isinstance(s, str) or not s for s in spaces)):
            raise ValueError('spaces must be a nonempty list of strings')
        params: dict[str, Any] = {'query': query, 'k': limit, 'include_own_pending': False,
                                  'include_provisional': include_provisional}
        if include_superseded:
            params['include_superseded'] = True
        if include_hidden:
            params['include_hidden'] = True
        if spaces is not None:
            params['spaces'] = spaces
        if audience is not None:
            params['audience'] = _text(audience, 'audience')
        if mode is not None:
            params['mode'] = _text(mode, 'mode')
        result = self._object(self._transport.call('dg.recall', params))
        hits = result.get('hits')
        if not isinstance(hits, list) or not all(isinstance(h, dict) for h in hits) or len(hits) > limit:
            raise GateError('invalid_response')
        return result

    # ---- proposals ----

    def remember(self, content: str | bytes, *, content_type: str, source_uri: str,
                 trust_class: str = 'T-1', space: str = 'main',
                 derives_from: Sequence[str] = ()) -> Proposal:
        """Propose one artifact. Bytes travel base64 and unchanged; the gate
        scans before it answers and holds the row until a reviewer that did
        not write it promotes it. Never retried; never carries an identity."""
        if not isinstance(content, (str, bytes)):
            raise ValueError('content must be text or bytes')
        for name, value in (('content_type', content_type), ('trust_class', trust_class),
                            ('source_uri', source_uri), ('space', space)):
            _text(value, name)
        try:
            raw = content.encode('utf-8') if isinstance(content, str) else content
        except UnicodeEncodeError:
            raise ValueError('invalid_utf8') from None
        params: dict[str, Any] = {'content': base64.b64encode(raw).decode('ascii'), 'encoding': 'base64',
                                  'content_type': content_type, 'trust_class': trust_class,
                                  'source_uri': source_uri, 'space': space}
        parents = _ids(derives_from, 'derives_from')
        if parents:
            params['derives_from'] = parents
        return Proposal.from_response(self._transport.call('dg.ingest', params))

    def learn(self, text: str, *, source_uri: str, kind: str = 'derived_fact',
              evidence: Sequence[str] = (), space: str = 'main', trust_class: str = 'T-5',
              replaces: str | None = None) -> Learning:
        """A learning: a claim distilled from evidence, proposed with its provenance.

        ``evidence`` becomes ``derives_from`` (never hides a parent, ADR-0035 d1).
        ``replaces`` names the claim this one should supersede: it joins the
        provenance and a note is left on the new row saying so, because
        only a gate decision replaces belief and ``supersedes`` is the
        operator's relation (ADR-0049 d5; GAPS 20). Default ``T-5``: a
        sentence a model wrote is a proposal about the world, not a fact.
        """
        if kind not in LEARNING_KINDS:
            raise ValueError(f'kind must be one of {sorted(LEARNING_KINDS)}')
        if not isinstance(text, str) or not text.strip():
            raise ValueError('a learning needs text')
        parents = _ids(evidence, 'evidence')
        if replaces is not None:
            _text(replaces, 'replaces')
            if replaces not in parents:
                parents.append(replaces)
        proposal = self.remember(text, content_type=kind, source_uri=source_uri, trust_class=trust_class,
                                 space=space, derives_from=parents)
        if replaces is not None and not proposal.duplicate:
            try:
                self.annotate(proposal.artifact_id, f'proposes: supersedes {replaces}', kind='note')
            except GateError as exc:
                if exc.kind not in ('forbidden_operation', 'writes_disabled', 'unsupported_operation'):
                    raise
        return Learning(proposal.artifact_id, proposal.state, proposal.duplicate, proposal.findings_count,
                        proposal.raw, kind, tuple(parents), replaces)

    def propose_skill(self, text: str, *, source_uri: str, space: str = 'main',
                      trust_class: str = 'T-5', derives_from: Sequence[str] = ()) -> Proposal:
        return self.remember(text, content_type='skill', source_uri=source_uri, trust_class=trust_class,
                             space=space, derives_from=derives_from)

    def propose_prompt_template(self, text: str, *, source_uri: str, space: str = 'main',
                                trust_class: str = 'T-5', derives_from: Sequence[str] = ()) -> Proposal:
        return self.remember(text, content_type='prompt_template', source_uri=source_uri,
                             trust_class=trust_class, space=space, derives_from=derives_from)

    def propose_bundle(self, archive: bytes, *, fetch_artifact_id: str, source_uri: str,
                       manifest: dict[str, Any] | None = None, space: str = 'main') -> dict[str, Any]:
        """A skill bundle: the archive is inspected here first (``skill_bundle``),
        then handed to ``dg.ingest_bundle`` beside the fetch record it derives
        from. The gate decides the class; no trust class travels."""
        from doublegate_sdk.skill_bundle import BundleError, inspect_zip
        if not isinstance(archive, bytes) or not archive:
            raise ValueError('archive must be nonempty bytes')
        _text(fetch_artifact_id, 'fetch_artifact_id'); _text(source_uri, 'source_uri'); _text(space, 'space')
        try:
            inspect_zip(archive)
        except BundleError as exc:
            raise ValueError(f'archive refused before the wire: {exc}') from None
        if not self.describe().dispatches('dg.ingest_bundle'):
            raise GateError('unsupported_operation', -32601, detail='this gate does not dispatch dg.ingest_bundle')
        params: dict[str, Any] = {'content': base64.b64encode(archive).decode('ascii'), 'source_uri': source_uri,
                                  'fetch_artifact_id': fetch_artifact_id, 'space': space}
        if manifest is not None:
            if not isinstance(manifest, dict):
                raise ValueError('manifest must be an object')
            params['manifest'] = manifest
        return self._object(self._transport.call('dg.ingest_bundle', params))

    # ---- annotations (ADR-0072): remarks, never decisions ----

    def annotate(self, artifact_id: str, body: str, *, kind: str = 'note',
                 reply_to: str | None = None) -> dict[str, Any]:
        if kind not in COMMENT_KINDS:
            raise ValueError(f'kind must be one of {sorted(COMMENT_KINDS)}')
        if kind == 'resolution' and not reply_to:
            raise ValueError('a resolution replies to a comment: reply_to is required')
        params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'body': _text(body, 'body'), 'kind': kind}
        if reply_to is not None:
            params['reply_to'] = _text(reply_to, 'reply_to')
        return self._object(self._transport.call('dg.comment', params))

    def raise_objection(self, artifact_id: str, body: str) -> dict[str, Any]:
        """An objection on the record. Only a human's objection holds promotion
        (ADR-0072); an agent's is a remark the reviewer will see (GAPS 21)."""
        return self.annotate(artifact_id, body, kind='objection')

    def resolve(self, artifact_id: str, body: str, *, reply_to: str) -> dict[str, Any]:
        return self.annotate(artifact_id, body, kind='resolution', reply_to=reply_to)

    def retract_comment(self, artifact_id: str, event_id: str, *, reason: str = '') -> dict[str, Any]:
        params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'event_id': _text(event_id, 'event_id')}
        if reason:
            params['reason'] = reason
        return self._object(self._transport.call('dg.comment_retract', params))

    def defer(self, artifact_id: str) -> dict[str, Any]:
        return self._one('dg.defer', artifact_id)

    # ---- serving for an audience (ADR-0052 d5: any identity but the writer's) ----

    def rank(self, artifact_id: str, *, audience: str, weight: float, reason: str = '') -> dict[str, Any]:
        if type(weight) not in (int, float) or not 0 < weight <= 10.0:
            raise ValueError('weight must be a number in (0, 10]')
        params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'),
                                  'audience': _text(audience, 'audience'), 'weight': float(weight)}
        if reason:
            params['reason'] = reason
        return self._object(self._transport.call('dg.rank', params))

    def hide(self, artifact_id: str, *, reason: str = '', audience: str = 'all', lift: bool = False) -> dict[str, Any]:
        if type(lift) is not bool:
            raise ValueError('lift must be boolean')
        if not lift and not reason:
            raise ValueError('a hide needs a reason; lift=True reverses one')
        params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'audience': _text(audience, 'audience')}
        if lift:
            params['lift'] = True
        else:
            params['reason'] = reason
        return self._object(self._transport.call('dg.hide', params))

    # ---- waiting on the gate ----

    def wait_for(self, artifact_id: str, states: Collection[str], *, timeout: float = 30.0,
                 interval: float = 0.5) -> dict[str, Any]:
        """Poll ``status`` until the row is in one of ``states`` or the deadline
        passes; returns the last row either way, with ``reached`` saying which."""
        wanted = frozenset(_text(s, 'states') for s in states)
        if not wanted:
            raise ValueError('states must name at least one state')
        if type(timeout) not in (int, float) or timeout < 0 or type(interval) not in (int, float) or interval <= 0:
            raise ValueError('timeout must be nonnegative and interval positive')
        deadline = time.monotonic() + timeout
        while True:
            row = self.status(artifact_id)
            if row.get('state') in wanted:
                return {**row, 'reached': True}
            if time.monotonic() >= deadline:
                return {**row, 'reached': False}
            time.sleep(min(interval, max(0.0, deadline - time.monotonic())))

    # ---- helpers ----

    def _one(self, method: str, artifact_id: str) -> dict[str, Any]:
        return self._object(self._transport.call(method, {'artifact_id': _text(artifact_id, 'artifact_id')}))

    @staticmethod
    def _object(result: Any) -> dict[str, Any]:
        if not isinstance(result, dict):
            raise GateError('invalid_response')
        return result

present

present() -> bool

Is a gate answering? Never raises; a refusal or an outage is False.

Source code in src/doublegate_sdk/knowledge.py

def present(self) -> bool:
    """Is a gate answering? Never raises; a refusal or an outage is ``False``."""
    try:
        answer = self._probe.call('dg.ping', {})
    except GateError:
        return False
    return isinstance(answer, dict) and bool(answer.get('pong'))

ping

ping() -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def ping(self) -> dict[str, Any]:
    return self._object(self._transport.call('dg.ping', {}))

describe

describe(*, refresh: bool = False) -> Capabilities

Source code in src/doublegate_sdk/knowledge.py

def describe(self, *, refresh: bool = False) -> Capabilities:
    if self._capabilities is None or refresh:
        self._capabilities = Capabilities.from_describe(self._object(self._transport.call('dg.describe', {})))
    return self._capabilities

status

status(artifact_id: str | None = None) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def status(self, artifact_id: str | None = None) -> dict[str, Any]:
    params = {} if artifact_id is None else {'artifact_id': _text(artifact_id, 'artifact_id')}
    return self._object(self._transport.call('dg.status', params))

why

why(artifact_id: str) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def why(self, artifact_id: str) -> dict[str, Any]:
    return self._one('dg.why', artifact_id)

tip

tip(artifact_id: str) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def tip(self, artifact_id: str) -> dict[str, Any]:
    return self._one('dg.tip', artifact_id)

relations

relations(artifact_id: str) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def relations(self, artifact_id: str) -> dict[str, Any]:
    return self._one('dg.relations', artifact_id)

approval

approval(artifact_id: str) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def approval(self, artifact_id: str) -> dict[str, Any]:
    return self._one('dg.approval', artifact_id)

node

node(artifact_id: str) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def node(self, artifact_id: str) -> dict[str, Any]:
    return self._one('dg.node', artifact_id)

comments

comments(artifact_id: str, *, include_retracted: bool = False) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def comments(self, artifact_id: str, *, include_retracted: bool = False) -> dict[str, Any]:
    if type(include_retracted) is not bool:
        raise ValueError('include_retracted must be boolean')
    return self._object(self._transport.call('dg.comments', {'artifact_id': _text(artifact_id, 'artifact_id'),
                                                             'include_retracted': include_retracted}))

pending

pending(*, limit: int = 100) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def pending(self, *, limit: int = 100) -> dict[str, Any]:
    if not _integer(limit, 1):
        raise ValueError('limit must be a positive integer')
    return self._object(self._transport.call('dg.pending', {'limit': limit}))

inventory

inventory(*, limit: int = 100, offset: int = 0, **filters: Any) -> InventoryPage

Source code in src/doublegate_sdk/knowledge.py

def inventory(self, *, limit: int = 100, offset: int = 0, **filters: Any) -> InventoryPage:
    if not _integer(limit, 1) or not _integer(offset, 0):
        raise ValueError('limit must be positive and offset nonnegative integers')
    if set(filters) - _FILTERS:
        raise ValueError('unsupported inventory parameter')
    result = self._transport.call('dg.inventory', {'limit': limit, 'offset': offset, **filters})
    return InventoryPage.from_response(self._object(result), requested_offset=offset)

inventory_pages

inventory_pages(*, limit: int = 100, max_pages: int = 10, **filters: Any) -> Iterator[InventoryPage]

Source code in src/doublegate_sdk/knowledge.py

def inventory_pages(self, *, limit: int = 100, max_pages: int = 10, **filters: Any) -> Iterator[InventoryPage]:
    if not _integer(max_pages, 1):
        raise ValueError('max_pages must be a positive integer')
    offset = 0
    for _ in range(max_pages):
        page = self.inventory(limit=limit, offset=offset, **filters)
        yield page
        if not page.has_more:
            return
        if page.next_offset is None:
            raise GateError('invalid_response')
        offset = page.next_offset
    raise GateError('page_limit')

recall

recall(query: str, *, limit: int = 10, spaces: list[str] | None = None, include_provisional: bool = False, include_superseded: bool = False, include_hidden: bool = False, audience: str | None = None, mode: str | None = None) -> dict[str, Any]

Served knowledge. The caller's own unreviewed rows are always excluded; provisional, superseded and hidden rows only on explicit request, and each hit keeps its flags — no score is relabelled as trust.

Source code in src/doublegate_sdk/knowledge.py

def recall(self, query: str, *, limit: int = 10, spaces: list[str] | None = None,
           include_provisional: bool = False, include_superseded: bool = False,
           include_hidden: bool = False, audience: str | None = None,
           mode: str | None = None) -> dict[str, Any]:
    """Served knowledge. The caller's own unreviewed rows are always excluded;
    provisional, superseded and hidden rows only on explicit request, and
    each hit keeps its flags — no score is relabelled as trust."""
    if not isinstance(query, str) or not query.strip() or not _integer(limit, 1):
        raise ValueError('nonempty query and positive integer limit required')
    for name, flag in (('include_provisional', include_provisional), ('include_superseded', include_superseded),
                       ('include_hidden', include_hidden)):
        if type(flag) is not bool:
            raise ValueError(f'{name} must be boolean')
    if spaces is not None and (not isinstance(spaces, list) or not spaces
                               or any(not isinstance(s, str) or not s for s in spaces)):
        raise ValueError('spaces must be a nonempty list of strings')
    params: dict[str, Any] = {'query': query, 'k': limit, 'include_own_pending': False,
                              'include_provisional': include_provisional}
    if include_superseded:
        params['include_superseded'] = True
    if include_hidden:
        params['include_hidden'] = True
    if spaces is not None:
        params['spaces'] = spaces
    if audience is not None:
        params['audience'] = _text(audience, 'audience')
    if mode is not None:
        params['mode'] = _text(mode, 'mode')
    result = self._object(self._transport.call('dg.recall', params))
    hits = result.get('hits')
    if not isinstance(hits, list) or not all(isinstance(h, dict) for h in hits) or len(hits) > limit:
        raise GateError('invalid_response')
    return result

remember

remember(content: str | bytes, *, content_type: str, source_uri: str, trust_class: str = 'T-1', space: str = 'main', derives_from: Sequence[str] = ()) -> Proposal

Propose one artifact. Bytes travel base64 and unchanged; the gate scans before it answers and holds the row until a reviewer that did not write it promotes it. Never retried; never carries an identity.

Source code in src/doublegate_sdk/knowledge.py

def remember(self, content: str | bytes, *, content_type: str, source_uri: str,
             trust_class: str = 'T-1', space: str = 'main',
             derives_from: Sequence[str] = ()) -> Proposal:
    """Propose one artifact. Bytes travel base64 and unchanged; the gate
    scans before it answers and holds the row until a reviewer that did
    not write it promotes it. Never retried; never carries an identity."""
    if not isinstance(content, (str, bytes)):
        raise ValueError('content must be text or bytes')
    for name, value in (('content_type', content_type), ('trust_class', trust_class),
                        ('source_uri', source_uri), ('space', space)):
        _text(value, name)
    try:
        raw = content.encode('utf-8') if isinstance(content, str) else content
    except UnicodeEncodeError:
        raise ValueError('invalid_utf8') from None
    params: dict[str, Any] = {'content': base64.b64encode(raw).decode('ascii'), 'encoding': 'base64',
                              'content_type': content_type, 'trust_class': trust_class,
                              'source_uri': source_uri, 'space': space}
    parents = _ids(derives_from, 'derives_from')
    if parents:
        params['derives_from'] = parents
    return Proposal.from_response(self._transport.call('dg.ingest', params))

learn

learn(text: str, *, source_uri: str, kind: str = 'derived_fact', evidence: Sequence[str] = (), space: str = 'main', trust_class: str = 'T-5', replaces: str | None = None) -> Learning

A learning: a claim distilled from evidence, proposed with its provenance.

evidence becomes derives_from (never hides a parent, ADR-0035 d1). replaces names the claim this one should supersede: it joins the provenance and a note is left on the new row saying so, because only a gate decision replaces belief and supersedes is the operator's relation (ADR-0049 d5; GAPS 20). Default T-5: a sentence a model wrote is a proposal about the world, not a fact.

Source code in src/doublegate_sdk/knowledge.py

def learn(self, text: str, *, source_uri: str, kind: str = 'derived_fact',
          evidence: Sequence[str] = (), space: str = 'main', trust_class: str = 'T-5',
          replaces: str | None = None) -> Learning:
    """A learning: a claim distilled from evidence, proposed with its provenance.

    ``evidence`` becomes ``derives_from`` (never hides a parent, ADR-0035 d1).
    ``replaces`` names the claim this one should supersede: it joins the
    provenance and a note is left on the new row saying so, because
    only a gate decision replaces belief and ``supersedes`` is the
    operator's relation (ADR-0049 d5; GAPS 20). Default ``T-5``: a
    sentence a model wrote is a proposal about the world, not a fact.
    """
    if kind not in LEARNING_KINDS:
        raise ValueError(f'kind must be one of {sorted(LEARNING_KINDS)}')
    if not isinstance(text, str) or not text.strip():
        raise ValueError('a learning needs text')
    parents = _ids(evidence, 'evidence')
    if replaces is not None:
        _text(replaces, 'replaces')
        if replaces not in parents:
            parents.append(replaces)
    proposal = self.remember(text, content_type=kind, source_uri=source_uri, trust_class=trust_class,
                             space=space, derives_from=parents)
    if replaces is not None and not proposal.duplicate:
        try:
            self.annotate(proposal.artifact_id, f'proposes: supersedes {replaces}', kind='note')
        except GateError as exc:
            if exc.kind not in ('forbidden_operation', 'writes_disabled', 'unsupported_operation'):
                raise
    return Learning(proposal.artifact_id, proposal.state, proposal.duplicate, proposal.findings_count,
                    proposal.raw, kind, tuple(parents), replaces)

propose_skill

propose_skill(text: str, *, source_uri: str, space: str = 'main', trust_class: str = 'T-5', derives_from: Sequence[str] = ()) -> Proposal

Source code in src/doublegate_sdk/knowledge.py

def propose_skill(self, text: str, *, source_uri: str, space: str = 'main',
                  trust_class: str = 'T-5', derives_from: Sequence[str] = ()) -> Proposal:
    return self.remember(text, content_type='skill', source_uri=source_uri, trust_class=trust_class,
                         space=space, derives_from=derives_from)

propose_prompt_template

propose_prompt_template(text: str, *, source_uri: str, space: str = 'main', trust_class: str = 'T-5', derives_from: Sequence[str] = ()) -> Proposal

Source code in src/doublegate_sdk/knowledge.py

def propose_prompt_template(self, text: str, *, source_uri: str, space: str = 'main',
                            trust_class: str = 'T-5', derives_from: Sequence[str] = ()) -> Proposal:
    return self.remember(text, content_type='prompt_template', source_uri=source_uri,
                         trust_class=trust_class, space=space, derives_from=derives_from)

propose_bundle

propose_bundle(archive: bytes, *, fetch_artifact_id: str, source_uri: str, manifest: dict[str, Any] | None = None, space: str = 'main') -> dict[str, Any]

A skill bundle: the archive is inspected here first (skill_bundle), then handed to dg.ingest_bundle beside the fetch record it derives from. The gate decides the class; no trust class travels.

Source code in src/doublegate_sdk/knowledge.py

def propose_bundle(self, archive: bytes, *, fetch_artifact_id: str, source_uri: str,
                   manifest: dict[str, Any] | None = None, space: str = 'main') -> dict[str, Any]:
    """A skill bundle: the archive is inspected here first (``skill_bundle``),
    then handed to ``dg.ingest_bundle`` beside the fetch record it derives
    from. The gate decides the class; no trust class travels."""
    from doublegate_sdk.skill_bundle import BundleError, inspect_zip
    if not isinstance(archive, bytes) or not archive:
        raise ValueError('archive must be nonempty bytes')
    _text(fetch_artifact_id, 'fetch_artifact_id'); _text(source_uri, 'source_uri'); _text(space, 'space')
    try:
        inspect_zip(archive)
    except BundleError as exc:
        raise ValueError(f'archive refused before the wire: {exc}') from None
    if not self.describe().dispatches('dg.ingest_bundle'):
        raise GateError('unsupported_operation', -32601, detail='this gate does not dispatch dg.ingest_bundle')
    params: dict[str, Any] = {'content': base64.b64encode(archive).decode('ascii'), 'source_uri': source_uri,
                              'fetch_artifact_id': fetch_artifact_id, 'space': space}
    if manifest is not None:
        if not isinstance(manifest, dict):
            raise ValueError('manifest must be an object')
        params['manifest'] = manifest
    return self._object(self._transport.call('dg.ingest_bundle', params))

annotate

annotate(artifact_id: str, body: str, *, kind: str = 'note', reply_to: str | None = None) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def annotate(self, artifact_id: str, body: str, *, kind: str = 'note',
             reply_to: str | None = None) -> dict[str, Any]:
    if kind not in COMMENT_KINDS:
        raise ValueError(f'kind must be one of {sorted(COMMENT_KINDS)}')
    if kind == 'resolution' and not reply_to:
        raise ValueError('a resolution replies to a comment: reply_to is required')
    params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'body': _text(body, 'body'), 'kind': kind}
    if reply_to is not None:
        params['reply_to'] = _text(reply_to, 'reply_to')
    return self._object(self._transport.call('dg.comment', params))

raise_objection

raise_objection(artifact_id: str, body: str) -> dict[str, Any]

An objection on the record. Only a human's objection holds promotion (ADR-0072); an agent's is a remark the reviewer will see (GAPS 21).

Source code in src/doublegate_sdk/knowledge.py

def raise_objection(self, artifact_id: str, body: str) -> dict[str, Any]:
    """An objection on the record. Only a human's objection holds promotion
    (ADR-0072); an agent's is a remark the reviewer will see (GAPS 21)."""
    return self.annotate(artifact_id, body, kind='objection')

resolve

resolve(artifact_id: str, body: str, *, reply_to: str) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def resolve(self, artifact_id: str, body: str, *, reply_to: str) -> dict[str, Any]:
    return self.annotate(artifact_id, body, kind='resolution', reply_to=reply_to)

retract_comment

retract_comment(artifact_id: str, event_id: str, *, reason: str = '') -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def retract_comment(self, artifact_id: str, event_id: str, *, reason: str = '') -> dict[str, Any]:
    params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'event_id': _text(event_id, 'event_id')}
    if reason:
        params['reason'] = reason
    return self._object(self._transport.call('dg.comment_retract', params))

defer

defer(artifact_id: str) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def defer(self, artifact_id: str) -> dict[str, Any]:
    return self._one('dg.defer', artifact_id)

rank

rank(artifact_id: str, *, audience: str, weight: float, reason: str = '') -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def rank(self, artifact_id: str, *, audience: str, weight: float, reason: str = '') -> dict[str, Any]:
    if type(weight) not in (int, float) or not 0 < weight <= 10.0:
        raise ValueError('weight must be a number in (0, 10]')
    params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'),
                              'audience': _text(audience, 'audience'), 'weight': float(weight)}
    if reason:
        params['reason'] = reason
    return self._object(self._transport.call('dg.rank', params))

hide

hide(artifact_id: str, *, reason: str = '', audience: str = 'all', lift: bool = False) -> dict[str, Any]

Source code in src/doublegate_sdk/knowledge.py

def hide(self, artifact_id: str, *, reason: str = '', audience: str = 'all', lift: bool = False) -> dict[str, Any]:
    if type(lift) is not bool:
        raise ValueError('lift must be boolean')
    if not lift and not reason:
        raise ValueError('a hide needs a reason; lift=True reverses one')
    params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'audience': _text(audience, 'audience')}
    if lift:
        params['lift'] = True
    else:
        params['reason'] = reason
    return self._object(self._transport.call('dg.hide', params))

wait_for

wait_for(artifact_id: str, states: Collection[str], *, timeout: float = 30.0, interval: float = 0.5) -> dict[str, Any]

Poll status until the row is in one of states or the deadline passes; returns the last row either way, with reached saying which.

Source code in src/doublegate_sdk/knowledge.py

def wait_for(self, artifact_id: str, states: Collection[str], *, timeout: float = 30.0,
             interval: float = 0.5) -> dict[str, Any]:
    """Poll ``status`` until the row is in one of ``states`` or the deadline
    passes; returns the last row either way, with ``reached`` saying which."""
    wanted = frozenset(_text(s, 'states') for s in states)
    if not wanted:
        raise ValueError('states must name at least one state')
    if type(timeout) not in (int, float) or timeout < 0 or type(interval) not in (int, float) or interval <= 0:
        raise ValueError('timeout must be nonnegative and interval positive')
    deadline = time.monotonic() + timeout
    while True:
        row = self.status(artifact_id)
        if row.get('state') in wanted:
            return {**row, 'reached': True}
        if time.monotonic() >= deadline:
            return {**row, 'reached': False}
        time.sleep(min(interval, max(0.0, deadline - time.monotonic())))

as_json

as_json(value: Any) -> str

Canonical text for a structured learning body: sorted keys, no NaN, unicode kept.

Source code in src/doublegate_sdk/knowledge.py

def as_json(value: Any) -> str:
    """Canonical text for a structured learning body: sorted keys, no NaN, unicode kept."""
    return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(',', ':'), allow_nan=False)

doublegate_sdk.curation

The curation client: what the operator does (ADR-0074 d2, d3).

Every decision here is the human's own signature: the transport must be curation-scoped and the client carries a ProofProvider that mints one single-use operator proof per verb (ADR-0028 d1). Admin scope on an organization gate is not operator standing (ADR-0040): without a provider the operator verbs are refused before I/O with proof_required.

RELATIONS module-attribute

RELATIONS: frozenset[str] = frozenset({'supersedes', 'restates', 'corrects', 'contradicts', 'extends'})

DECISIONS module-attribute

DECISIONS: frozenset[str] = frozenset({'promote', 'reject', 'hold', 'clear', 'archive'})

CurationClient

The operator's view of a gate: decisions, relations, serving policy, keys.

Source code in src/doublegate_sdk/curation.py

class CurationClient:
    """The operator's view of a gate: decisions, relations, serving policy, keys."""

    def __init__(self, transport: GateTransport, proof: ProofProvider | None = None):
        scope = getattr(transport, 'scope', CURATION)
        if scope != CURATION:
            raise ValueError('the curation client needs a curation-scoped transport')
        self._transport = transport
        self._proof = proof
        #: the same gate, read the agent's way
        self.knowledge = _Reader(transport)

    def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
        op = OPERATIONS[method]
        if op.proof == 'operator':
            if self._proof is None:
                raise GateError('proof_required', detail=f'{method} is the operator\'s: supply a ProofProvider')
            params = {**params, 'operator_proof': self._proof.prove(self._transport)}
        elif op.proof == 'optional' and self._proof is not None:
            params = {**params, 'operator_proof': self._proof.prove(self._transport)}
        result = self._transport.call(method, params)
        if not isinstance(result, dict):
            raise GateError('invalid_response')
        return result

    # ---- decisions (ADR-0073: approval is a sign, veto is a demote) ----

    def sign(self, artifact_id: str, decision: str, *, note: str = '', promote: bool = False) -> dict[str, Any]:
        if decision not in DECISIONS:
            raise ValueError(f'decision must be one of {sorted(DECISIONS)}')
        params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'decision': decision}
        if note:
            params['note'] = note
        if promote:
            params['promote'] = True
        return self._call('dg.sign', params)

    def approve(self, artifact_id: str, *, note: str = '') -> dict[str, Any]:
        """Sign ``promote`` and promote under one proof; a promotion the boundary
        refuses is reported in ``promote.refused`` and the signature stands."""
        return self.sign(artifact_id, 'promote', note=note, promote=True)

    def hold(self, artifact_id: str, *, note: str = '') -> dict[str, Any]:
        return self.sign(artifact_id, 'hold', note=note)

    def clear_finding(self, artifact_id: str, *, note: str = '') -> dict[str, Any]:
        return self.sign(artifact_id, 'clear', note=note)

    def reject(self, artifact_id: str, reason: str, *, in_favour_of: str | None = None) -> dict[str, Any]:
        params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'reason': _text(reason, 'reason')}
        if in_favour_of is not None:
            params['in_favour_of'] = _text(in_favour_of, 'in_favour_of')
        return self._call('dg.reject', params)

    def veto(self, artifact_id: str, reason: str) -> dict[str, Any]:
        """The human veto of an admitted row: ``dg.demote`` (ADR-0073). Ends the
        claim; keeps the content."""
        return self._call('dg.demote', {'artifact_id': _text(artifact_id, 'artifact_id'), 'reason': _text(reason, 'reason')})

    def promote(self, artifact_id: str, *, importance: float | None = None, supersedes: str | None = None) -> dict[str, Any]:
        params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id')}
        if importance is not None:
            if type(importance) not in (int, float) or not 0 <= importance <= 1:
                raise ValueError('importance must be in 0..1')
            params['importance'] = float(importance)
        if supersedes is not None:
            params['supersedes'] = _text(supersedes, 'supersedes')
        return self._call('dg.promote', params)

    # ---- the lifecycle passes an operator runs by hand ----

    def scan(self, artifact_id: str) -> dict[str, Any]:
        return self._call('dg.scan', {'artifact_id': _text(artifact_id, 'artifact_id')})

    def grade(self, artifact_id: str) -> dict[str, Any]:
        """One model call per stance on one row; the gate engine, not this client, judges."""
        return self._call('dg.grade', {'artifact_id': _text(artifact_id, 'artifact_id')})

    def grade_pending(self, *, limit: int = 50) -> dict[str, Any]:
        return self._call('dg.grade_pending', {'limit': int(limit)})

    def settle_pending(self) -> dict[str, Any]:
        """Promote what quorum already settled; a row under an open human objection stays (ADR-0072)."""
        return self._call('dg.settle_pending', {})

    def semantic_review(self, artifact_id: str, *, settle: bool = False) -> dict[str, Any]:
        return self._call('dg.semantic_review', {'artifact_id': _text(artifact_id, 'artifact_id'), 'settle': bool(settle)})

    # ---- relations (ADR-0035): only supersedes hides ----

    def relate(self, child: str, kind: str, parent: str) -> dict[str, Any]:
        if kind not in RELATIONS:
            raise ValueError(f'kind must be one of {sorted(RELATIONS)}')
        return self._call('dg.relate', {'artifact_id': _text(child, 'child'), 'kind': kind, 'parent': _text(parent, 'parent')})

    def supersede(self, new: str, old: str) -> dict[str, Any]:
        return self.relate(new, 'supersedes', old)

    def override_tip(self, artifact_id: str, *, reason: str = '') -> dict[str, Any]:
        params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id')}
        if reason:
            params['reason'] = reason
        return self._call('dg.override_tip', params)

    # ---- serving policy (ADR-0052) ----

    def rank(self, artifact_id: str, *, audience: str, weight: float, reason: str = '') -> dict[str, Any]:
        if type(weight) not in (int, float) or not 0 < weight <= 10.0:
            raise ValueError('weight must be a number in (0, 10]')
        params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'audience': _text(audience, 'audience'),
                                  'weight': float(weight)}
        if reason:
            params['reason'] = reason
        return self._call('dg.rank', params)

    def hide(self, artifact_id: str, *, reason: str = '', audience: str = 'all', lift: bool = False) -> dict[str, Any]:
        if not lift and not reason:
            raise ValueError('a hide needs a reason; lift=True reverses one')
        params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'audience': _text(audience, 'audience')}
        if lift:
            params['lift'] = True
        else:
            params['reason'] = reason
        return self._call('dg.hide', params)

    def ban(self, kind: str, subject: str, *, reason: str = '', until_ms: int | None = None, lift: bool = False) -> dict[str, Any]:
        if kind not in ('writer', 'source'):
            raise ValueError("kind must be 'writer' or 'source'")
        if not lift and not reason:
            raise ValueError('a ban needs a reason; lift=True reverses one')
        params: dict[str, Any] = {'kind': kind, 'subject': _text(subject, 'subject')}
        if lift:
            params['lift'] = True
        else:
            params['reason'] = reason
            if until_ms is not None:
                if type(until_ms) is not int or until_ms <= 0:
                    raise ValueError('until_ms must be a positive integer')
                params['until'] = until_ms
        return self._call('dg.ban', params)

    def bans(self) -> dict[str, Any]:
        return self._call('dg.bans', {})

    # ---- the operator's presence and the handoff ----

    def away(self, *, until: str | None = None, note: str = '') -> dict[str, Any]:
        params: dict[str, Any] = {}
        if until is not None:
            params['until'] = _text(until, 'until')
        if note:
            params['note'] = note
        return self._call('dg.away', params)

    def back(self) -> dict[str, Any]:
        return self._call('dg.back', {})

    def defer(self, artifact_id: str) -> dict[str, Any]:
        return self._call('dg.defer', {'artifact_id': _text(artifact_id, 'artifact_id')})

    def submit(self, artifact_id: str) -> dict[str, Any]:
        return self._call('dg.submit', {'artifact_id': _text(artifact_id, 'artifact_id')})

    def submit_drain(self) -> dict[str, Any]:
        return self._call('dg.submit_drain', {})

    def review(self, *, cap: int | None = None) -> dict[str, Any]:
        return self._call('dg.review', {} if cap is None else {'cap': int(cap)})

    # ---- the organization gate's keys (ADR-0033, ADR-0040) ----

    def keys(self, action: str = 'list', **fields: Any) -> dict[str, Any]:
        """``dg.keys``: list | show | issue | revoke. Issue and revoke are the
        operator's; a client gate answers ``unsupported_operation``."""
        if action not in ('list', 'show', 'issue', 'revoke'):
            raise ValueError('action must be list | show | issue | revoke')
        params = {'action': action, **fields}
        if action in ('issue', 'revoke'):
            if self._proof is None:
                raise GateError('proof_required', detail='issuing or revoking a key is the operator\'s (ADR-0040)')
            params['operator_proof'] = self._proof.prove(self._transport)
        result = self._transport.call('dg.keys', params)
        if not isinstance(result, dict):
            raise GateError('invalid_response')
        return result

knowledge instance-attribute

knowledge = _Reader(transport)

sign

sign(artifact_id: str, decision: str, *, note: str = '', promote: bool = False) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def sign(self, artifact_id: str, decision: str, *, note: str = '', promote: bool = False) -> dict[str, Any]:
    if decision not in DECISIONS:
        raise ValueError(f'decision must be one of {sorted(DECISIONS)}')
    params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'decision': decision}
    if note:
        params['note'] = note
    if promote:
        params['promote'] = True
    return self._call('dg.sign', params)

approve

approve(artifact_id: str, *, note: str = '') -> dict[str, Any]

Sign promote and promote under one proof; a promotion the boundary refuses is reported in promote.refused and the signature stands.

Source code in src/doublegate_sdk/curation.py

def approve(self, artifact_id: str, *, note: str = '') -> dict[str, Any]:
    """Sign ``promote`` and promote under one proof; a promotion the boundary
    refuses is reported in ``promote.refused`` and the signature stands."""
    return self.sign(artifact_id, 'promote', note=note, promote=True)

hold

hold(artifact_id: str, *, note: str = '') -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def hold(self, artifact_id: str, *, note: str = '') -> dict[str, Any]:
    return self.sign(artifact_id, 'hold', note=note)

clear_finding

clear_finding(artifact_id: str, *, note: str = '') -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def clear_finding(self, artifact_id: str, *, note: str = '') -> dict[str, Any]:
    return self.sign(artifact_id, 'clear', note=note)

reject

reject(artifact_id: str, reason: str, *, in_favour_of: str | None = None) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def reject(self, artifact_id: str, reason: str, *, in_favour_of: str | None = None) -> dict[str, Any]:
    params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'reason': _text(reason, 'reason')}
    if in_favour_of is not None:
        params['in_favour_of'] = _text(in_favour_of, 'in_favour_of')
    return self._call('dg.reject', params)

veto

veto(artifact_id: str, reason: str) -> dict[str, Any]

The human veto of an admitted row: dg.demote (ADR-0073). Ends the claim; keeps the content.

Source code in src/doublegate_sdk/curation.py

def veto(self, artifact_id: str, reason: str) -> dict[str, Any]:
    """The human veto of an admitted row: ``dg.demote`` (ADR-0073). Ends the
    claim; keeps the content."""
    return self._call('dg.demote', {'artifact_id': _text(artifact_id, 'artifact_id'), 'reason': _text(reason, 'reason')})

promote

promote(artifact_id: str, *, importance: float | None = None, supersedes: str | None = None) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def promote(self, artifact_id: str, *, importance: float | None = None, supersedes: str | None = None) -> dict[str, Any]:
    params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id')}
    if importance is not None:
        if type(importance) not in (int, float) or not 0 <= importance <= 1:
            raise ValueError('importance must be in 0..1')
        params['importance'] = float(importance)
    if supersedes is not None:
        params['supersedes'] = _text(supersedes, 'supersedes')
    return self._call('dg.promote', params)

scan

scan(artifact_id: str) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def scan(self, artifact_id: str) -> dict[str, Any]:
    return self._call('dg.scan', {'artifact_id': _text(artifact_id, 'artifact_id')})

grade

grade(artifact_id: str) -> dict[str, Any]

One model call per stance on one row; the gate engine, not this client, judges.

Source code in src/doublegate_sdk/curation.py

def grade(self, artifact_id: str) -> dict[str, Any]:
    """One model call per stance on one row; the gate engine, not this client, judges."""
    return self._call('dg.grade', {'artifact_id': _text(artifact_id, 'artifact_id')})

grade_pending

grade_pending(*, limit: int = 50) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def grade_pending(self, *, limit: int = 50) -> dict[str, Any]:
    return self._call('dg.grade_pending', {'limit': int(limit)})

settle_pending

settle_pending() -> dict[str, Any]

Promote what quorum already settled; a row under an open human objection stays (ADR-0072).

Source code in src/doublegate_sdk/curation.py

def settle_pending(self) -> dict[str, Any]:
    """Promote what quorum already settled; a row under an open human objection stays (ADR-0072)."""
    return self._call('dg.settle_pending', {})

semantic_review

semantic_review(artifact_id: str, *, settle: bool = False) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def semantic_review(self, artifact_id: str, *, settle: bool = False) -> dict[str, Any]:
    return self._call('dg.semantic_review', {'artifact_id': _text(artifact_id, 'artifact_id'), 'settle': bool(settle)})

relate

relate(child: str, kind: str, parent: str) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def relate(self, child: str, kind: str, parent: str) -> dict[str, Any]:
    if kind not in RELATIONS:
        raise ValueError(f'kind must be one of {sorted(RELATIONS)}')
    return self._call('dg.relate', {'artifact_id': _text(child, 'child'), 'kind': kind, 'parent': _text(parent, 'parent')})

supersede

supersede(new: str, old: str) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def supersede(self, new: str, old: str) -> dict[str, Any]:
    return self.relate(new, 'supersedes', old)

override_tip

override_tip(artifact_id: str, *, reason: str = '') -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def override_tip(self, artifact_id: str, *, reason: str = '') -> dict[str, Any]:
    params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id')}
    if reason:
        params['reason'] = reason
    return self._call('dg.override_tip', params)

rank

rank(artifact_id: str, *, audience: str, weight: float, reason: str = '') -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def rank(self, artifact_id: str, *, audience: str, weight: float, reason: str = '') -> dict[str, Any]:
    if type(weight) not in (int, float) or not 0 < weight <= 10.0:
        raise ValueError('weight must be a number in (0, 10]')
    params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'audience': _text(audience, 'audience'),
                              'weight': float(weight)}
    if reason:
        params['reason'] = reason
    return self._call('dg.rank', params)

hide

hide(artifact_id: str, *, reason: str = '', audience: str = 'all', lift: bool = False) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def hide(self, artifact_id: str, *, reason: str = '', audience: str = 'all', lift: bool = False) -> dict[str, Any]:
    if not lift and not reason:
        raise ValueError('a hide needs a reason; lift=True reverses one')
    params: dict[str, Any] = {'artifact_id': _text(artifact_id, 'artifact_id'), 'audience': _text(audience, 'audience')}
    if lift:
        params['lift'] = True
    else:
        params['reason'] = reason
    return self._call('dg.hide', params)

ban

ban(kind: str, subject: str, *, reason: str = '', until_ms: int | None = None, lift: bool = False) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def ban(self, kind: str, subject: str, *, reason: str = '', until_ms: int | None = None, lift: bool = False) -> dict[str, Any]:
    if kind not in ('writer', 'source'):
        raise ValueError("kind must be 'writer' or 'source'")
    if not lift and not reason:
        raise ValueError('a ban needs a reason; lift=True reverses one')
    params: dict[str, Any] = {'kind': kind, 'subject': _text(subject, 'subject')}
    if lift:
        params['lift'] = True
    else:
        params['reason'] = reason
        if until_ms is not None:
            if type(until_ms) is not int or until_ms <= 0:
                raise ValueError('until_ms must be a positive integer')
            params['until'] = until_ms
    return self._call('dg.ban', params)

bans

bans() -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def bans(self) -> dict[str, Any]:
    return self._call('dg.bans', {})

away

away(*, until: str | None = None, note: str = '') -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def away(self, *, until: str | None = None, note: str = '') -> dict[str, Any]:
    params: dict[str, Any] = {}
    if until is not None:
        params['until'] = _text(until, 'until')
    if note:
        params['note'] = note
    return self._call('dg.away', params)

back

back() -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def back(self) -> dict[str, Any]:
    return self._call('dg.back', {})

defer

defer(artifact_id: str) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def defer(self, artifact_id: str) -> dict[str, Any]:
    return self._call('dg.defer', {'artifact_id': _text(artifact_id, 'artifact_id')})

submit

submit(artifact_id: str) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def submit(self, artifact_id: str) -> dict[str, Any]:
    return self._call('dg.submit', {'artifact_id': _text(artifact_id, 'artifact_id')})

submit_drain

submit_drain() -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def submit_drain(self) -> dict[str, Any]:
    return self._call('dg.submit_drain', {})

review

review(*, cap: int | None = None) -> dict[str, Any]

Source code in src/doublegate_sdk/curation.py

def review(self, *, cap: int | None = None) -> dict[str, Any]:
    return self._call('dg.review', {} if cap is None else {'cap': int(cap)})

keys

keys(action: str = 'list', **fields: Any) -> dict[str, Any]

dg.keys: list | show | issue | revoke. Issue and revoke are the operator's; a client gate answers unsupported_operation.

Source code in src/doublegate_sdk/curation.py

def keys(self, action: str = 'list', **fields: Any) -> dict[str, Any]:
    """``dg.keys``: list | show | issue | revoke. Issue and revoke are the
    operator's; a client gate answers ``unsupported_operation``."""
    if action not in ('list', 'show', 'issue', 'revoke'):
        raise ValueError('action must be list | show | issue | revoke')
    params = {'action': action, **fields}
    if action in ('issue', 'revoke'):
        if self._proof is None:
            raise GateError('proof_required', detail='issuing or revoking a key is the operator\'s (ADR-0040)')
        params['operator_proof'] = self._proof.prove(self._transport)
    result = self._transport.call('dg.keys', params)
    if not isinstance(result, dict):
        raise GateError('invalid_response')
    return result

doublegate_sdk.transport

Two transports, one framing, one allowlist (ADR-0074 d1, d6).

No connection on import or construction; no discovery, credentials, retries or alternate endpoints. The transport's scope is the widest class of verb it will emit — checked before any I/O — and a refusal here is a courtesy: the receiving gate refuses regardless, from the identity it derived off the connection (I2). A caller-supplied GateTransport owns its own allowlist.

Both transports here carry dg.* verbs: UnixSocketTransport over the daemon socket, HttpTransport over POST /rpc. The third door, POST /mcp (the MCP tool catalog), is doublegate_sdk.client.HttpMcpTransport behind doublegate_sdk.client.McpTransport; same shape, different vocabulary.

MAX_REQUEST_BYTES module-attribute

MAX_REQUEST_BYTES = 1048576

DEFAULT_MAX_RESPONSE_BYTES module-attribute

DEFAULT_MAX_RESPONSE_BYTES = 1048576

GateTransport

Bases: Protocol

Explicit operation transport; the receiver remains authoritative.

Source code in src/doublegate_sdk/transport.py

class GateTransport(Protocol):
    """Explicit operation transport; the receiver remains authoritative."""
    def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...

call

call(method: str, params: dict[str, Any]) -> dict[str, Any]

Source code in src/doublegate_sdk/transport.py

def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...

UnixSocketTransport

Bases: _Bounded

Bounded local JSON-RPC over the gate's own socket, newline-framed.

Source code in src/doublegate_sdk/transport.py

class UnixSocketTransport(_Bounded):
    """Bounded local JSON-RPC over the gate's own socket, newline-framed."""

    def __init__(self, path: str | os.PathLike[str], *, timeout: float = 10.0,
                 max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES, scope: str = READ):
        self._init_bounds(timeout, max_response_bytes, scope)
        raw_path = os.fspath(path)
        if not isinstance(raw_path, str) or not raw_path or '\x00' in raw_path:
            raise ValueError('path must be nonempty text without NUL characters')
        self._path = os.path.abspath(raw_path)

    @property
    def path(self) -> str:
        return self._path

    def with_timeout(self, timeout: float) -> UnixSocketTransport:
        return UnixSocketTransport(self._path, timeout=timeout, max_response_bytes=self._max_bytes, scope=self._scope)

    def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
        writing = self._admit(method)
        if not hasattr(socket, 'AF_UNIX'):
            raise GateError('unsupported_transport')
        request = encode_request(method, params) + b'\n'
        deadline = time.monotonic() + self._timeout
        sending = False
        try:
            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
                connection.settimeout(_remaining(deadline))
                connection.connect(self._path)
                connection.settimeout(_remaining(deadline))
                sending = True
                connection.sendall(request)
                response = bytearray()
                while b'\n' not in response:
                    connection.settimeout(_remaining(deadline))
                    chunk = connection.recv(min(65536, self._max_bytes + 1 - len(response)))
                    if not chunk:
                        raise GateError('invalid_response')
                    response.extend(chunk)
                    if len(response) > self._max_bytes:
                        raise GateError('response_too_large')
        except TimeoutError:
            raise GateError('timeout', outcome_unknown=writing and sending) from None
        except OSError:
            raise GateError('unavailable', outcome_unknown=writing and sending) from None
        except GateError as exc:
            exc.outcome_unknown = writing and sending
            raise
        return decode_response(bytes(response), writing=writing)

path property

path: str

with_timeout

with_timeout(timeout: float) -> UnixSocketTransport

Source code in src/doublegate_sdk/transport.py

def with_timeout(self, timeout: float) -> UnixSocketTransport:
    return UnixSocketTransport(self._path, timeout=timeout, max_response_bytes=self._max_bytes, scope=self._scope)

call

call(method: str, params: dict[str, Any]) -> dict[str, Any]

Source code in src/doublegate_sdk/transport.py

def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
    writing = self._admit(method)
    if not hasattr(socket, 'AF_UNIX'):
        raise GateError('unsupported_transport')
    request = encode_request(method, params) + b'\n'
    deadline = time.monotonic() + self._timeout
    sending = False
    try:
        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
            connection.settimeout(_remaining(deadline))
            connection.connect(self._path)
            connection.settimeout(_remaining(deadline))
            sending = True
            connection.sendall(request)
            response = bytearray()
            while b'\n' not in response:
                connection.settimeout(_remaining(deadline))
                chunk = connection.recv(min(65536, self._max_bytes + 1 - len(response)))
                if not chunk:
                    raise GateError('invalid_response')
                response.extend(chunk)
                if len(response) > self._max_bytes:
                    raise GateError('response_too_large')
    except TimeoutError:
        raise GateError('timeout', outcome_unknown=writing and sending) from None
    except OSError:
        raise GateError('unavailable', outcome_unknown=writing and sending) from None
    except GateError as exc:
        exc.outcome_unknown = writing and sending
        raise
    return decode_response(bytes(response), writing=writing)

HttpTransport

Bases: _Bounded

The same message over POST {base}/rpc with a bearer: the client gate's console token on its loopback listener, an organization gate's admin key, or an assertion an agent mints from its own key (ADR-0080 d4). bearer is the string itself or a zero-argument callable returning the current one, called once per request — the shape a short-lived assertion needs, re-minted by its holder before it expires. JSON-RPC refusals ride a 200; the door's own refusals are HTTP statuses (401 auth, 403 scope, 413, 429/503 with Retry-After).

Source code in src/doublegate_sdk/transport.py

class HttpTransport(_Bounded):
    """The same message over ``POST {base}/rpc`` with a bearer: the client
    gate's console token on its loopback listener, an organization gate's
    admin key, or an assertion an agent mints from its own key (ADR-0080 d4).
    ``bearer`` is the string itself or a zero-argument callable returning the
    current one, called once per request — the shape a short-lived assertion
    needs, re-minted by its holder before it expires. JSON-RPC refusals ride a
    200; the door's own refusals are HTTP statuses (401 auth, 403 scope, 413,
    429/503 with Retry-After)."""

    def __init__(self, base_url: str, *, bearer: str | Callable[[], str] = '', timeout: float = 10.0,
                 max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES, scope: str = READ):
        self._init_bounds(timeout, max_response_bytes, scope)
        if not isinstance(base_url, str) or not base_url:
            raise ValueError('base_url must be nonempty text')
        parts = urllib.parse.urlsplit(base_url)
        if parts.scheme not in ('http', 'https') or not parts.hostname:
            raise ValueError('base_url must be an http(s) URL with a host')
        if not callable(bearer):
            _bearer_text(bearer)
        self._parts, self._bearer = parts, bearer
        self._base_url = base_url

    @property
    def base_url(self) -> str:
        return self._base_url

    def with_timeout(self, timeout: float) -> HttpTransport:
        return HttpTransport(self._base_url, bearer=self._bearer, timeout=timeout,
                             max_response_bytes=self._max_bytes, scope=self._scope)

    def _connection(self, timeout: float) -> http.client.HTTPConnection:
        cls = http.client.HTTPSConnection if self._parts.scheme == 'https' else http.client.HTTPConnection
        port = self._parts.port or (443 if self._parts.scheme == 'https' else 80)
        return cls(self._parts.hostname or '', port, timeout=timeout)

    def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
        writing = self._admit(method)
        body = encode_request(method, params)
        headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
        bearer = _bearer_text(self._bearer()) if callable(self._bearer) else self._bearer
        if bearer:
            headers['Authorization'] = 'Bearer ' + bearer
        path = (self._parts.path.rstrip('/') or '') + '/rpc'
        sending = False
        try:
            connection = self._connection(self._timeout)
            try:
                sending = True
                connection.request('POST', path, body=body, headers=headers)
                response = connection.getresponse()
                status = response.status
                retry_after = response.getheader('Retry-After')
                raw = response.read(self._max_bytes + 1)
            finally:
                connection.close()
        except TimeoutError:
            raise GateError('timeout', outcome_unknown=writing and sending) from None
        except (OSError, http.client.HTTPException):
            raise GateError('unavailable', outcome_unknown=writing and sending) from None
        if len(raw) > self._max_bytes:
            raise GateError('response_too_large', outcome_unknown=writing)
        if status != 200:
            retry = None
            if retry_after is not None:
                try:
                    retry = int(float(retry_after) * 1000)
                except ValueError:
                    retry = None
            raise GateError(kind_of_status(status), retry_after_ms=retry,
                            outcome_unknown=writing and status >= 500,
                            detail=raw.decode('utf-8', 'replace') if raw else None)
        return decode_response(raw, writing=writing)

base_url property

base_url: str

with_timeout

with_timeout(timeout: float) -> HttpTransport

Source code in src/doublegate_sdk/transport.py

def with_timeout(self, timeout: float) -> HttpTransport:
    return HttpTransport(self._base_url, bearer=self._bearer, timeout=timeout,
                         max_response_bytes=self._max_bytes, scope=self._scope)

call

call(method: str, params: dict[str, Any]) -> dict[str, Any]

Source code in src/doublegate_sdk/transport.py

def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
    writing = self._admit(method)
    body = encode_request(method, params)
    headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
    bearer = _bearer_text(self._bearer()) if callable(self._bearer) else self._bearer
    if bearer:
        headers['Authorization'] = 'Bearer ' + bearer
    path = (self._parts.path.rstrip('/') or '') + '/rpc'
    sending = False
    try:
        connection = self._connection(self._timeout)
        try:
            sending = True
            connection.request('POST', path, body=body, headers=headers)
            response = connection.getresponse()
            status = response.status
            retry_after = response.getheader('Retry-After')
            raw = response.read(self._max_bytes + 1)
        finally:
            connection.close()
    except TimeoutError:
        raise GateError('timeout', outcome_unknown=writing and sending) from None
    except (OSError, http.client.HTTPException):
        raise GateError('unavailable', outcome_unknown=writing and sending) from None
    if len(raw) > self._max_bytes:
        raise GateError('response_too_large', outcome_unknown=writing)
    if status != 200:
        retry = None
        if retry_after is not None:
            try:
                retry = int(float(retry_after) * 1000)
            except ValueError:
                retry = None
        raise GateError(kind_of_status(status), retry_after_ms=retry,
                        outcome_unknown=writing and status >= 500,
                        detail=raw.decode('utf-8', 'replace') if raw else None)
    return decode_response(raw, writing=writing)

encode_request

encode_request(method: str, params: dict[str, Any]) -> bytes

One JSON-RPC 2.0 message, id 1, bounded.

Source code in src/doublegate_sdk/transport.py

def encode_request(method: str, params: dict[str, Any]) -> bytes:
    """One JSON-RPC 2.0 message, id 1, bounded."""
    if not isinstance(params, dict):
        raise ValueError('params must be an object')
    request = json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': method, 'params': params},
                         allow_nan=False).encode()
    if len(request) > MAX_REQUEST_BYTES:
        raise GateError('request_too_large')
    return request

decode_response

decode_response(raw: bytes, *, writing: bool) -> dict[str, Any]

The result of a JSON-RPC reply, or the refusal it carried as a GateError.

Source code in src/doublegate_sdk/transport.py

def decode_response(raw: bytes, *, writing: bool) -> dict[str, Any]:
    """The result of a JSON-RPC reply, or the refusal it carried as a ``GateError``."""
    try:
        data = json.loads(raw, object_pairs_hook=_unique_object)
        if not (isinstance(data, dict) and data.get('jsonrpc') == '2.0'):
            raise GateError('invalid_response')
        if not (type(data.get('id')) is int and data['id'] == 1):
            raise GateError('invalid_response')
        if ('result' in data) == ('error' in data):
            raise GateError('invalid_response')
        if 'error' in data:
            err = data['error']
            code = err['code']
            if type(code) is not int:
                raise GateError('invalid_response')
            extra = err.get('data') if isinstance(err.get('data'), dict) else {}
            retry = extra.get('retry_after_ms')
            raise GateError(kind_of_code(code), code,
                            retry_after_ms=retry if type(retry) is int else None,
                            outcome_unknown=writing and code == -32603,
                            detail=err.get('message') if isinstance(err.get('message'), str) else None)
        if not isinstance(data['result'], dict):
            raise GateError('invalid_response')
        return data['result']
    except (ValueError, KeyError, TypeError, AttributeError):
        raise GateError('invalid_response', outcome_unknown=writing) from None
    except GateError as exc:
        if exc.kind == 'invalid_response':
            exc.outcome_unknown = writing
        raise

doublegate_sdk.proof

The operator proof seam (ADR-0074 d3).

An operator verb carries {nonce, sig}: a fresh dg.challenge nonce and the operator key's signature over it (ADR-0028 d1). The SDK never holds that key. It asks a ProofProvider — a caller-supplied signer, or the client daemon's own dg.operator_proof on the host where the key lives — and forwards what it gets (ADR-0040: wrappers forward proof, they do not synthesize it). Nonces are single-use: one proof per verb, minted just before the call.

OperatorProof

Bases: TypedDict

Source code in src/doublegate_sdk/proof.py

class OperatorProof(TypedDict):
    nonce: str
    sig: str

nonce instance-attribute

nonce: str

sig instance-attribute

sig: str

ProofProvider

Bases: Protocol

Source code in src/doublegate_sdk/proof.py

class ProofProvider(Protocol):
    def prove(self, transport: Any) -> OperatorProof: ...

prove

prove(transport: Any) -> OperatorProof

Source code in src/doublegate_sdk/proof.py

def prove(self, transport: Any) -> OperatorProof: ...

SignerProof

Fetch a nonce from the gate and sign it with the caller's function (bytes -> bytes, a detached Ed25519 signature). The key stays with the caller.

Source code in src/doublegate_sdk/proof.py

class SignerProof:
    """Fetch a nonce from the gate and sign it with the caller's function
    (``bytes -> bytes``, a detached Ed25519 signature). The key stays with the caller."""

    def __init__(self, sign: Callable[[bytes], bytes]):
        if not callable(sign):
            raise ValueError('sign must be callable')
        self._sign = sign

    def prove(self, transport: Any) -> OperatorProof:
        challenge = transport.call('dg.challenge', {})
        nonce = challenge.get('nonce') if isinstance(challenge, dict) else None
        if not isinstance(nonce, str) or not nonce:
            raise GateError('invalid_response')
        sig = self._sign(nonce.encode('ascii'))
        if not isinstance(sig, (bytes, bytearray)) or not sig:
            raise GateError('proof_required', detail='the signer returned no signature')
        return {'nonce': nonce, 'sig': bytes(sig).hex()}

prove

prove(transport: Any) -> OperatorProof

Source code in src/doublegate_sdk/proof.py

def prove(self, transport: Any) -> OperatorProof:
    challenge = transport.call('dg.challenge', {})
    nonce = challenge.get('nonce') if isinstance(challenge, dict) else None
    if not isinstance(nonce, str) or not nonce:
        raise GateError('invalid_response')
    sig = self._sign(nonce.encode('ascii'))
    if not isinstance(sig, (bytes, bytearray)) or not sig:
        raise GateError('proof_required', detail='the signer returned no signature')
    return {'nonce': nonce, 'sig': bytes(sig).hex()}

ServerMinted

dg.operator_proof: the client daemon signs the nonce with the key on its host. Presence on that host is the whole authorization; a gate whose key is unreadable, or an organization gate, refuses and this says so.

Source code in src/doublegate_sdk/proof.py

class ServerMinted:
    """``dg.operator_proof``: the client daemon signs the nonce with the key on
    its host. Presence on that host is the whole authorization; a gate whose
    key is unreadable, or an organization gate, refuses and this says so."""

    def prove(self, transport: Any) -> OperatorProof:
        try:
            answer = transport.call('dg.operator_proof', {})
        except GateError as exc:
            if exc.kind in ('invalid_params', 'unsupported_operation', 'refused'):
                raise GateError('proof_required', exc.code, detail=exc.detail) from None
            raise
        proof = answer.get('operator_proof', answer) if isinstance(answer, dict) else None
        if not isinstance(proof, dict) or not isinstance(proof.get('nonce'), str) or not isinstance(proof.get('sig'), str):
            raise GateError('invalid_response')
        return {'nonce': proof['nonce'], 'sig': proof['sig']}

prove

prove(transport: Any) -> OperatorProof

Source code in src/doublegate_sdk/proof.py

def prove(self, transport: Any) -> OperatorProof:
    try:
        answer = transport.call('dg.operator_proof', {})
    except GateError as exc:
        if exc.kind in ('invalid_params', 'unsupported_operation', 'refused'):
            raise GateError('proof_required', exc.code, detail=exc.detail) from None
        raise
    proof = answer.get('operator_proof', answer) if isinstance(answer, dict) else None
    if not isinstance(proof, dict) or not isinstance(proof.get('nonce'), str) or not isinstance(proof.get('sig'), str):
        raise GateError('invalid_response')
    return {'nonce': proof['nonce'], 'sig': proof['sig']}

doublegate_sdk.capabilities

What a running gate says it answers (dg.describe), parsed once.

The catalog is the schema (ADR-0044): a client learns a verb's presence, its parameters and whether this build dispatches it from the gate itself, never from prose. content_types comes from the structured enum on dg.ingest's content_type parameter; a gate that does not publish one yields an empty set, and a caller must not read that as "accepts anything".

VerbInfo dataclass

Source code in src/doublegate_sdk/capabilities.py

@dataclass(frozen=True, slots=True)
class VerbInfo:
    name: str
    params: tuple[str, ...]
    dispatched: bool
    proof: str
    role: str

name instance-attribute

name: str

params instance-attribute

params: tuple[str, ...]

dispatched instance-attribute

dispatched: bool

proof instance-attribute

proof: str

role instance-attribute

role: str

Capabilities dataclass

Source code in src/doublegate_sdk/capabilities.py

@dataclass(frozen=True)
class Capabilities:
    role: str
    version: str
    verbs: dict[str, VerbInfo]
    content_types: frozenset[str]

    @classmethod
    def from_describe(cls, data: dict[str, Any]) -> Capabilities:
        if not isinstance(data, dict) or not isinstance(data.get('verbs'), list):
            raise ValueError('not a dg.describe answer')
        verbs: dict[str, VerbInfo] = {}
        content_types: set[str] = set()
        for verb in data['verbs']:
            if not isinstance(verb, dict) or not isinstance(verb.get('name'), str):
                continue
            params = verb.get('params') if isinstance(verb.get('params'), list) else []
            names = tuple(p['name'] for p in params if isinstance(p, dict) and isinstance(p.get('name'), str))
            verbs[verb['name']] = VerbInfo(verb['name'], names, bool(verb.get('dispatched', True)),
                                           str(verb.get('proof', 'none')), str(verb.get('role', 'both')))
            if verb['name'] == 'dg.ingest':
                for p in params:
                    if isinstance(p, dict) and p.get('name') == 'content_type' and isinstance(p.get('enum'), list):
                        content_types.update(t for t in p['enum'] if isinstance(t, str))
        return cls(str(data.get('role', 'any')), str(data.get('version', '')), verbs, frozenset(content_types))

    def dispatches(self, verb: str) -> bool:
        info = self.verbs.get(verb)
        return info is not None and info.dispatched

    def accepts_content_type(self, content_type: str) -> bool:
        return content_type in self.content_types

    def params_of(self, verb: str) -> tuple[str, ...]:
        info = self.verbs.get(verb)
        return info.params if info else ()

role instance-attribute

role: str

version instance-attribute

version: str

verbs instance-attribute

verbs: dict[str, VerbInfo]

content_types instance-attribute

content_types: frozenset[str]

from_describe classmethod

from_describe(data: dict[str, Any]) -> Capabilities

Source code in src/doublegate_sdk/capabilities.py

@classmethod
def from_describe(cls, data: dict[str, Any]) -> Capabilities:
    if not isinstance(data, dict) or not isinstance(data.get('verbs'), list):
        raise ValueError('not a dg.describe answer')
    verbs: dict[str, VerbInfo] = {}
    content_types: set[str] = set()
    for verb in data['verbs']:
        if not isinstance(verb, dict) or not isinstance(verb.get('name'), str):
            continue
        params = verb.get('params') if isinstance(verb.get('params'), list) else []
        names = tuple(p['name'] for p in params if isinstance(p, dict) and isinstance(p.get('name'), str))
        verbs[verb['name']] = VerbInfo(verb['name'], names, bool(verb.get('dispatched', True)),
                                       str(verb.get('proof', 'none')), str(verb.get('role', 'both')))
        if verb['name'] == 'dg.ingest':
            for p in params:
                if isinstance(p, dict) and p.get('name') == 'content_type' and isinstance(p.get('enum'), list):
                    content_types.update(t for t in p['enum'] if isinstance(t, str))
    return cls(str(data.get('role', 'any')), str(data.get('version', '')), verbs, frozenset(content_types))

dispatches

dispatches(verb: str) -> bool

Source code in src/doublegate_sdk/capabilities.py

def dispatches(self, verb: str) -> bool:
    info = self.verbs.get(verb)
    return info is not None and info.dispatched

accepts_content_type

accepts_content_type(content_type: str) -> bool

Source code in src/doublegate_sdk/capabilities.py

def accepts_content_type(self, content_type: str) -> bool:
    return content_type in self.content_types

params_of

params_of(verb: str) -> tuple[str, ...]

Source code in src/doublegate_sdk/capabilities.py

def params_of(self, verb: str) -> tuple[str, ...]:
    info = self.verbs.get(verb)
    return info.params if info else ()

doublegate_sdk.errors

One error base, one gate vocabulary (ADR-0074 d5).

DoublegateError is the base every SDK failure shares, so a caller can catch the SDK without swallowing unrelated application exceptions. GateError is the one raised on a wire: kind is what a caller acts on; code is the server's JSON-RPC number or HTTP status when there was one; detail carries the server's message, capped, and never enters str(exc) so a log line cannot leak what a gate said about a held row. The two tables here are the same ones the constellation app's rpc.ts uses; change both or neither.

DETAIL_MAX module-attribute

DETAIL_MAX = 512

CODE_KINDS module-attribute

CODE_KINDS: dict[int, str] = {-32000: 'identity', -32001: 'request_too_large', -32002: 'request_too_large', -32600: 'invalid_params', -32602: 'invalid_params', -32601: 'unsupported_operation', -32006: 'busy', -32009: 'banned', -32010: 'refused', -32012: 'refused', -32013: 'refused', -32014: 'refused'}

HTTP_KINDS module-attribute

HTTP_KINDS: dict[int, str] = {401: 'auth', 403: 'scope', 413: 'request_too_large', 429: 'busy', 503: 'busy'}

KINDS module-attribute

KINDS: frozenset[str] = frozenset({'unavailable', 'timeout', 'unsupported_transport', 'invalid_response', 'request_too_large', 'response_too_large', 'writes_disabled', 'forbidden_operation', 'page_limit', 'proof_required', 'redirect_refused', 'identity', 'invalid_params', 'unsupported_operation', 'busy', 'banned', 'refused', 'remote_error', 'auth', 'scope', 'unauthorized', 'forbidden', 'tool_error'})

DoublegateError

Bases: Exception

Catch SDK failures without swallowing unrelated application exceptions.

Concrete exceptions retain their own safe diagnostic, protocol code and established ValueError/RuntimeError compatibility. kind names the category; specialized fields provide details without parsing human-readable text.

Source code in src/doublegate_sdk/errors.py

class DoublegateError(Exception):
    """Catch SDK failures without swallowing unrelated application exceptions.

    Concrete exceptions retain their own safe diagnostic, protocol code and
    established ValueError/RuntimeError compatibility. ``kind`` names the category;
    specialized fields provide details without parsing human-readable text.
    """

    kind = 'sdk_error'

kind class-attribute instance-attribute

kind = 'sdk_error'

GateError

Bases: DoublegateError, RuntimeError

A fixed diagnostic. str(exc) is the kind (and code); the server's text is detail.

Source code in src/doublegate_sdk/errors.py

class GateError(DoublegateError, RuntimeError):
    """A fixed diagnostic. ``str(exc)`` is the kind (and code); the server's text is ``detail``."""

    def __init__(self, kind: str, code: int | None = None, *, retry_after_ms: int | None = None,
                 outcome_unknown: bool = False, detail: str | None = None):
        if kind not in KINDS:
            raise ValueError(f'unknown error kind {kind!r}')
        self.kind, self.code = kind, code
        self.retry_after_ms = retry_after_ms
        self.outcome_unknown = outcome_unknown
        self.detail = None if detail is None else str(detail)[:DETAIL_MAX]
        super().__init__(kind if code is None else f'{kind} ({code})')

    def __repr__(self) -> str:
        return f'GateError(kind={self.kind!r}, code={self.code!r}, outcome_unknown={self.outcome_unknown!r})'

retry_after_ms instance-attribute

retry_after_ms = retry_after_ms

outcome_unknown instance-attribute

outcome_unknown = outcome_unknown

detail instance-attribute

detail = None if detail is None else str(detail)[:DETAIL_MAX]

kind_of_code

kind_of_code(code: int) -> str

Source code in src/doublegate_sdk/errors.py

def kind_of_code(code: int) -> str:
    return CODE_KINDS.get(code, 'remote_error')

kind_of_status

kind_of_status(status: int) -> str

Source code in src/doublegate_sdk/errors.py

def kind_of_status(status: int) -> str:
    return HTTP_KINDS.get(status, 'remote_error')

doublegate_sdk.operations

The operation table: every dg.* verb of both gate catalogs, with what a call needs (ADR-0074 d1). The catalog is the schema (I7); this table is its client-side reading and the allowlist every transport enforces before I/O.

scope is the widest client that may emit the verb:

• read — answers questions, changes nothing;
• knowledge — what an agent may do: propose, annotate, weight for an audience; never decide (the gate is not callable by the thing being gated);
• curation — the operator's verbs, most of them under an operator proof.

SCOPES module-attribute

SCOPES: tuple[str, ...] = (READ, KNOWLEDGE, CURATION)

OPERATIONS module-attribute

OPERATIONS: dict[str, Operation] = {op.rpc: op for op in (_op('dg.ping', mutates=False, scope=READ), _op('dg.describe', mutates=False, scope=READ), _op('dg.status', mutates=False, scope=READ), _op('dg.pending', mutates=False, scope=READ), _op('dg.inventory', mutates=False, scope=READ), _op('dg.recall', mutates=False, scope=READ), _op('dg.why', mutates=False, scope=READ), _op('dg.audit', mutates=False, scope=READ), _op('dg.tip', mutates=False, scope=READ), _op('dg.relations', mutates=False, scope=READ), _op('dg.approval', mutates=False, scope=READ), _op('dg.graph', mutates=False, scope=READ), _op('dg.node', mutates=False, scope=READ), _op('dg.comments', mutates=False, scope=READ), _op('dg.review', mutates=False, scope=READ), _op('dg.review_runs', mutates=False, scope=READ), _op('dg.engagement', mutates=False, scope=READ), _op('dg.bans', mutates=False, scope=READ), _op('dg.settings', mutates=False, scope=READ), _op('dg.verify', mutates=False, scope=READ), _op('dg.challenge', mutates=False, scope=READ), _op('dg.ingest', mutates=True, scope=KNOWLEDGE), _op('dg.ingest_bundle', mutates=True, scope=KNOWLEDGE), _op('dg.comment', mutates=True, proof='optional', scope=KNOWLEDGE), _op('dg.comment_retract', mutates=True, proof='optional', scope=KNOWLEDGE), _op('dg.defer', mutates=True, scope=KNOWLEDGE), _op('dg.rank', mutates=True, proof='optional', scope=KNOWLEDGE), _op('dg.hide', mutates=True, proof='optional', scope=KNOWLEDGE), _op('dg.crawler_policy', mutates=True, proof='operator', scope=KNOWLEDGE), _op('dg.operator_proof', mutates=False, proof='presence', scope=CURATION), _op('dg.sign', mutates=True, proof='operator', scope=CURATION), _op('dg.promote', mutates=True, proof='operator', scope=CURATION), _op('dg.demote', mutates=True, proof='operator', scope=CURATION), _op('dg.reject', mutates=True, proof='operator', scope=CURATION), _op('dg.relate', mutates=True, proof='operator', scope=CURATION), _op('dg.override_tip', mutates=True, proof='operator', scope=CURATION), _op('dg.ban', mutates=True, proof='operator', scope=CURATION), _op('dg.away', mutates=True, proof='operator', scope=CURATION), _op('dg.back', mutates=True, proof='operator', scope=CURATION), _op('dg.semantic_review', mutates=True, scope=CURATION), _op('dg.scan', mutates=True, scope=CURATION), _op('dg.scan_pending', mutates=True, scope=CURATION), _op('dg.grade', mutates=True, scope=CURATION), _op('dg.grade_pending', mutates=True, scope=CURATION), _op('dg.review_pending', mutates=True, scope=CURATION), _op('dg.settle_pending', mutates=True, scope=CURATION), _op('dg.submit', mutates=True, role='client', scope=CURATION), _op('dg.submit_drain', mutates=True, role='client', scope=CURATION), _op('dg.publish', mutates=True, proof='presence', scope=CURATION), _op('dg.reconcile', mutates=True, proof='presence', scope=CURATION), _op('dg.rebuild', mutates=True, proof='presence', scope=CURATION), _op('dg.settings_validate', mutates=False, scope=CURATION), _op('dg.settings_save', mutates=True, scope=CURATION), _op('dg.collection', mutates=False, role='org', scope=READ), _op('dg.policy', mutates=False, role='org', scope=READ), _op('dg.countersign_status', mutates=False, role='org', scope=READ), _op('dg.audit_chain', mutates=False, role='org', scope=READ), _op('dg.org_review', mutates=False, role='org', scope=READ), _op('dg.keys', mutates=True, proof='operator', role='org', scope=CURATION), _op('dg.work', mutates=True, proof='presence', role='org', scope=CURATION))}

Operation dataclass

Source code in src/doublegate_sdk/operations.py

@dataclass(frozen=True, slots=True)
class Operation:
    rpc: str
    mutates: bool
    proof: str      # none | presence | optional | operator
    role: str       # client | org | both
    scope: str      # read | knowledge | curation

rpc instance-attribute

rpc: str

mutates instance-attribute

mutates: bool

proof instance-attribute

proof: str

role instance-attribute

role: str

scope instance-attribute

scope: str

allows

allows(scope: str, rpc: str) -> bool

True when a transport of scope may emit rpc; unknown verbs are never allowed.

Source code in src/doublegate_sdk/operations.py

def allows(scope: str, rpc: str) -> bool:
    """True when a transport of ``scope`` may emit ``rpc``; unknown verbs are never allowed."""
    op = OPERATIONS.get(rpc)
    return op is not None and _RANK[op.scope] <= _RANK[scope]

in_scope

in_scope(scope: str) -> frozenset[str]

Source code in src/doublegate_sdk/operations.py

def in_scope(scope: str) -> frozenset[str]:
    return frozenset(rpc for rpc in OPERATIONS if allows(scope, rpc))

check_scope

check_scope(scope: str) -> str

Source code in src/doublegate_sdk/operations.py

def check_scope(scope: str) -> str:
    if scope not in SCOPES:
        raise ValueError(f"scope must be one of {SCOPES}, got {scope!r}")
    return scope

doublegate_sdk.inventory

The inventory page as the gate answers it: metadata, scope and coverage, never converted into a claim of admission.

InventoryPage dataclass

Source code in src/doublegate_sdk/inventory.py

@dataclass(frozen=True)
class InventoryPage:
    records: tuple[dict[str, Any], ...]
    total: int
    limit: int
    offset: int
    has_more: bool
    next_offset: int | None
    complete: bool
    coverage_errors: tuple[Any, ...]
    scope: dict[str, Any]
    counts: dict[str, Any]

    @classmethod
    def from_response(cls, data: dict[str, Any], *, requested_offset: int) -> InventoryPage:
        try:
            rows, coverage = data['page'], data['coverage']
            _require(isinstance(rows, list) and all(isinstance(r, dict) for r in rows))
            _require(_integer(data['total'], 0) and _integer(data['limit'], 1))
            _require(_integer(data['offset'], 0) and data['offset'] == requested_offset)
            _require(len(rows) <= data['limit'] and len(rows) <= data['total'])
            _require(type(data['has_more']) is bool)
            _require(data['has_more'] == (data['offset'] + len(rows) < data['total']))
            _require(not rows or data['offset'] + len(rows) <= data['total'])
            following = data['next_offset']
            if data['has_more']:
                _require(bool(rows) and _integer(following, data['offset'] + 1))
                _require(following == data['offset'] + len(rows))
            else:
                _require(following is None)
            _require(type(coverage['complete']) is bool and isinstance(coverage['errors'], list))
            _require(not coverage['complete'] or not coverage['errors'])
            _require(isinstance(data['scope'], dict) and isinstance(data['counts'], dict))
            return cls(tuple(dict(r) for r in rows), data['total'], data['limit'],
                       data['offset'], data['has_more'], following, coverage['complete'],
                       tuple(coverage['errors']), dict(data['scope']), dict(data['counts']))
        except (KeyError, TypeError, AssertionError):
            raise GateError('invalid_response') from None

records instance-attribute

records: tuple[dict[str, Any], ...]

total instance-attribute

total: int

limit instance-attribute

limit: int

offset instance-attribute

offset: int

has_more instance-attribute

has_more: bool

next_offset instance-attribute

next_offset: int | None

complete instance-attribute

complete: bool

coverage_errors instance-attribute

coverage_errors: tuple[Any, ...]

scope instance-attribute

scope: dict[str, Any]

counts instance-attribute

counts: dict[str, Any]

from_response classmethod

from_response(data: dict[str, Any], *, requested_offset: int) -> InventoryPage

Source code in src/doublegate_sdk/inventory.py

@classmethod
def from_response(cls, data: dict[str, Any], *, requested_offset: int) -> InventoryPage:
    try:
        rows, coverage = data['page'], data['coverage']
        _require(isinstance(rows, list) and all(isinstance(r, dict) for r in rows))
        _require(_integer(data['total'], 0) and _integer(data['limit'], 1))
        _require(_integer(data['offset'], 0) and data['offset'] == requested_offset)
        _require(len(rows) <= data['limit'] and len(rows) <= data['total'])
        _require(type(data['has_more']) is bool)
        _require(data['has_more'] == (data['offset'] + len(rows) < data['total']))
        _require(not rows or data['offset'] + len(rows) <= data['total'])
        following = data['next_offset']
        if data['has_more']:
            _require(bool(rows) and _integer(following, data['offset'] + 1))
            _require(following == data['offset'] + len(rows))
        else:
            _require(following is None)
        _require(type(coverage['complete']) is bool and isinstance(coverage['errors'], list))
        _require(not coverage['complete'] or not coverage['errors'])
        _require(isinstance(data['scope'], dict) and isinstance(data['counts'], dict))
        return cls(tuple(dict(r) for r in rows), data['total'], data['limit'],
                   data['offset'], data['has_more'], following, coverage['complete'],
                   tuple(coverage['errors']), dict(data['scope']), dict(data['counts']))
    except (KeyError, TypeError, AssertionError):
        raise GateError('invalid_response') from None

doublegate_sdk.org_rest

The organization gate's two REST doors a fleet client speaks (ADR-0074 d6): PUT /submissions/{artifact_id} (ADR-0030) and GET /outcomes (ADR-0041). The body a submission carries is doublegate_sdk.submission's; this module only moves it and classifies what the door said. Policy — what to do about a retry, where to keep a cursor — is the caller's.

SubmissionOutcome dataclass

What the door said about one PUT, in the four words the outbox acts on: accepted (202/200), retry (503/429 with the door's delay, or any unexpected status), stopped (401/403: a key problem does not become correct by waiting), refused (400/413/422: this body, not the key).

Source code in src/doublegate_sdk/org_rest.py

@dataclass(frozen=True)
class SubmissionOutcome:
    """What the door said about one PUT, in the four words the outbox acts on:
    ``accepted`` (202/200), ``retry`` (503/429 with the door's delay, or any
    unexpected status), ``stopped`` (401/403: a key problem does not become
    correct by waiting), ``refused`` (400/413/422: this body, not the key)."""
    result: str
    status_code: int | None
    retry_after_s: float | None
    reason: str

result instance-attribute

result: str

status_code instance-attribute

status_code: int | None

retry_after_s instance-attribute

retry_after_s: float | None

reason instance-attribute

reason: str

OutcomesPage dataclass

Source code in src/doublegate_sdk/org_rest.py

@dataclass(frozen=True)
class OutcomesPage:
    events: tuple[dict[str, Any], ...]
    next: int | None

events instance-attribute

events: tuple[dict[str, Any], ...]

next instance-attribute

next: int | None

reason_of

reason_of(body: bytes) -> str

The door's error field, or a short prefix of the body.

Source code in src/doublegate_sdk/org_rest.py

def reason_of(body: bytes) -> str:
    """The door's ``error`` field, or a short prefix of the body."""
    try:
        data = json.loads(body)
        if isinstance(data, dict) and isinstance(data.get('error'), str):
            return data['error'][:200]
    except ValueError:
        pass
    return body[:200].decode('utf-8', 'replace')

classify_submission_status

classify_submission_status(code: int, retry_after: str | None, body: bytes) -> SubmissionOutcome

Source code in src/doublegate_sdk/org_rest.py

def classify_submission_status(code: int, retry_after: str | None, body: bytes) -> SubmissionOutcome:
    if code in (202, 200):
        return SubmissionOutcome(ACCEPTED, code, None, '')
    if code in (503, 429):
        delay: float | None
        try:
            delay = float(retry_after) if retry_after is not None else None
        except ValueError:
            delay = None
        return SubmissionOutcome(RETRY, code, delay, 'backpressure')
    if code in (401, 403):
        return SubmissionOutcome(STOPPED, code, None, f'authz: HTTP {code} — check the token')
    if code in (400, 413, 422):
        return SubmissionOutcome(REFUSED, code, None, f'door: HTTP {code} — {reason_of(body)}')
    return SubmissionOutcome(RETRY, code, None, f'unexpected HTTP {code}')

put_submission

put_submission(endpoint: str, artifact_id: str, body: bytes, token: str, *, timeout_s: float = 10.0) -> SubmissionOutcome

One PUT. A transport failure raises GateError('unavailable'); the caller decides that it is retryable, because that is a policy question.

Source code in src/doublegate_sdk/org_rest.py

def put_submission(endpoint: str, artifact_id: str, body: bytes, token: str, *,
                   timeout_s: float = 10.0) -> SubmissionOutcome:
    """One PUT. A transport failure raises ``GateError('unavailable')``; the
    caller decides that it is retryable, because that is a policy question."""
    if not isinstance(token, str) or not token or any(c.isspace() for c in token):
        raise ValueError('token must be nonempty text without whitespace')
    conn, base = _connection(endpoint, timeout_s)
    try:
        conn.request('PUT', f'{base}/submissions/{artifact_id}', body=body,
                     headers={'Content-Type': 'application/json', 'Authorization': f'Bearer {token}'})
        resp = conn.getresponse()
        return classify_submission_status(resp.status, resp.getheader('Retry-After'), resp.read())
    except (OSError, http.client.HTTPException) as exc:
        raise GateError('unavailable', outcome_unknown=True, detail=f'transport: {exc}') from None
    finally:
        conn.close()

pull_outcomes

pull_outcomes(endpoint: str, token: str, *, since: int, limit: int = 500, timeout_s: float = 10.0) -> OutcomesPage

GET /outcomes?since=<cursor>&limit=<n>. auth/scope on 401/403, remote_error on any other non-200, invalid_response on a non-JSON body.

Source code in src/doublegate_sdk/org_rest.py

def pull_outcomes(endpoint: str, token: str, *, since: int, limit: int = 500,
                  timeout_s: float = 10.0) -> OutcomesPage:
    """``GET /outcomes?since=<cursor>&limit=<n>``. ``auth``/``scope`` on 401/403,
    ``remote_error`` on any other non-200, ``invalid_response`` on a non-JSON body."""
    if type(since) is not int or since < 0 or type(limit) is not int or limit < 1:
        raise ValueError('since must be a nonnegative and limit a positive integer')
    conn, base = _connection(endpoint, timeout_s)
    try:
        conn.request('GET', f'{base}/outcomes?since={since}&limit={limit}',
                     headers={'Accept': 'application/json', 'Authorization': f'Bearer {token}'})
        resp = conn.getresponse()
        code, data = resp.status, resp.read()
    except (OSError, http.client.HTTPException) as exc:
        raise GateError('unavailable', detail=f'transport: {exc}') from None
    finally:
        conn.close()
    if code in (401, 403):
        raise GateError('auth' if code == 401 else 'scope', detail=reason_of(data))
    if code != 200:
        raise GateError('remote_error', detail=f'HTTP {code}')
    try:
        body = json.loads(data)
    except ValueError:
        raise GateError('invalid_response') from None
    events = body.get('events') if isinstance(body, dict) else None
    if not isinstance(events, list) or not all(isinstance(e, dict) for e in events):
        raise GateError('invalid_response')
    nxt = body.get('next')
    return OutcomesPage(tuple(events), nxt if type(nxt) is int else None)