semiont

The Python SDK for Semiont, an open platform for building trusted AI knowledge bases: a shared workspace where humans and AI agents annotate, connect and govern a corpus of documents.

This package is the client. With it a program reads and writes one knowledge base: it adds resources, annotates and links them, searches them, gathers context for a model, and hears what the other participants do as they do it. A person's application and an AI agent use the same client. The SDK does not tell them apart.

It has the same namespaces, methods and live queries as the TypeScript SDK and the Rust SDK, and is held to the same conformance suite. New to Semiont? The Introduction explains the domain and the ideas the API falls out of. Its code is TypeScript, and the ideas are this package's too.

Install

pip install semiont

It requires Python 3.12 or later, and runs on asyncio. It is checked by mypy --strict and by pyright in strict mode, and ships its types (py.typed): a wrong id, a wrong payload for a channel and an unhandled code or state are errors a type checker reports.

One package holds all of it: the client, the transport that carries it over HTTP to a knowledge base's gateway, signing in, and the doubles a test is built on. Its dependencies are httpx, pydantic and opentelemetry-api.

A client speaks to a running knowledge base. The Quick Start sets one up on your own machine.

Signing in

A transport sends the token it is given. Where the token comes from is one of three things, and each keeps it renewed: half its lifetime before it expires, at most five minutes before, and at once when the gateway refuses it.

An agent is a service account's token exchanged at the gateway for the token of the agent (provider, model). It is what a service holds: a transport, its agent's token, and the channels it awaits replies on. The gateway answers the exchange with the agent's DID, which is who its work is attributed to: agent.did, once it is signed in, as the gateway minted it.

from semiont.bus import reply_channels_for
from semiont.http import AgentToken, Credential, HttpTransport, ServiceToken
from semiont.operations import JOB_CLAIM
from semiont.watched import reached


async def work(gateway: str, issuer: str, secret: str) -> None:
    service = ServiceToken(Credential(issuer=issuer, client_id="semiont-smelter", client_secret=secret))
    async with (
        AgentToken(gateway, provider="ollama", model="gemma2:27b", service=service) as agent,
        HttpTransport(gateway, token=agent.token, refresher=agent.refresh, channels=reply_channels_for(JOB_CLAIM)) as transport,
    ):
        print(f"working as {agent.did}")  # who its work is attributed to, as the gateway named it
        print(await reached(transport.state, lambda state: state == "open"))  # its stream is open, as the agent

A person is signed in at the issuer the knowledge base trusts, and the sign-in is kept. semiont login keeps one for each stack, and a session here is built over it; a script on a machine with no launcher makes its own by the device grant, which shows a code to approve in any browser. No password passes through this process.

from pathlib import Path

from semiont.http import DeviceCode, session_from_kept, sign_in_device
from semiont.sign_in_store import FILE_NAME, SignInStore, state_dir, this_system


def show(code: DeviceCode) -> None:
    print(f"Open {code.verification_uri} and enter {code.user_code}")


async def as_me(gateway: str, home: str) -> None:
    directory = state_dir(this_system(), home=home, xdg_state_home=None, local_app_data=None)
    if directory is None:
        return
    kept = SignInStore(Path(directory) / FILE_NAME).entry("local")
    if await kept.held() is None:
        await sign_in_device(gateway, kept, show)
    async with session_from_kept(gateway, kb_id="local", kept=kept) as session:
        print(session.user.value, (await session.client.system.status()).version)

• An agent whose renewal fails keeps the token it has and tries again: a process holding a token that still works does not stop over one bad round trip. Signing in is the one step that raises.
• A person's session ends once, and says why. A token the issuer will not renew ends it as expired, and a token the issuer has just issued and the gateway refuses ends it as refused (on_auth_failed). The sign-in is then forgotten, so a dead credential is never sent again. What a session does at its start and when it is refused is specs/src/session/cases.json, row for row.
• A renewal spends its refresh token once. Renewals asked for together are one request of the issuer, a renewal the issuer did not refuse is tried again inside a bounded budget, and one under way runs to its end whoever stops waiting for it.
• The store is the launcher's (specs/src/sign-in-store): the same file and the same lock, on macOS, Linux and Windows. A sign-in made here serves the launcher's verbs, and one it renewed serves this session. MemorySignIn keeps one for the life of the process instead, and sign_out revokes one at its issuer and forgets it.
• SEMIONT_BUS_LOG is the only environment variable the package reads (Telemetry). Everything else it needs, it is given: an application reads its own environment (HOME, a secret, an address) and says what it found.

Eight verbs

Every operation belongs to one of eight flows: verbs for what a participant does with a shared corpus. Four write, three read, and one directs attention. Each is a namespace of the client.

yield is Python's own word, so its namespace is yield_. Beside the eight are job, auth and system. What each flow means is in docs/protocol/flows.

A first program

From a signed-in client, this ingests a paper, has a model mark the concepts it mentions, gathers the context around it, and generates a summary grounded in that context. Those are the Quick Start's last two steps, from code, and the two that come next. Signing in is how a client comes to be signed in.

from semiont.client import SemiontClient
from semiont.identifiers import ResourceId
from semiont.transport import PutBinaryRequest, Transport
from semiont.types import GenerationJobParams, JobDeclinedResult, LinkingJobParams


async def summarize(client: SemiontClient[Transport], paper: bytes) -> ResourceId | None:
    # Ingest: the paper's bytes become a resource.
    created = await client.yield_.resource(
        PutBinaryRequest(
            name="Attention Is All You Need",
            file=paper,
            format="application/pdf",
            storage_uri="file://papers/attention-is-all-you-need.pdf",
        )
    )
    paper_id = created.resource_id

    # Annotate: a model reads it and marks each mention of a concept.
    await client.mark.delegate(paper_id, LinkingJobParams(motivation="linking", entity_types=["Concept"]))

    # Gather: the paper, its annotations, and what the knowledge base holds around it.
    context = await client.gather.resource(paper_id)

    # Generate: a new resource, grounded in that context and linked to its source.
    done = await client.yield_.delegate(
        GenerationJobParams(
            title="Attention Is All You Need: a summary",
            storage_uri="file://generated/attention-summary.md",
            context=context,
            task="summary",
        )
    )
    # What a `yield` job reports is the resource it made. One that could not read what it was given declines, and makes nothing.
    made = done.result
    return None if made is None or isinstance(made, JobDeclinedResult) else made.resource_id

Both resources, and every annotation the model made, are in the knowledge base for the next participant, person or agent, to read and build on.

Where to go from here:

• The Developer Guide has each of these steps as a recipe, and the ones after them: reading, searching, annotating by hand, reacting to what others do, and testing. Its code is TypeScript, and the calls are this package's too.
• The agent skills are whole scripts, one per task, for an AI coding assistant to load: ingesting a corpus, annotating it, linking it, and the layers built on those. They are TypeScript as well.

The client

A client is eleven namespaces over one transport: the eight verbs, one per flow of the protocol, and job, auth and system beside them. Each method is a row of specs/src/client/surface.json, the table every Semiont SDK is held to. HttpTransport is the transport this package ships: it reaches a knowledge base's gateway, and sends the token it is given.

from semiont.client import SemiontClient
from semiont.http import HttpTransport
from semiont.identifiers import AnnotationId, ResourceId
from semiont.namespaces.follow import JobAttemptFailed, JobCompleted, JobCreated, JobProgressed
from semiont.transport import Transport
from semiont.types import HighlightingJobParams
from semiont.watched import Variable


async def annotate(client: SemiontClient[Transport], resource: ResourceId, annotation: AnnotationId) -> None:
    described = await client.browse.resource(resource).fresh()  # a query, read once
    text = await client.browse.resource_content(resource)  # asked once, answered once
    print(described.name, len(text))

    async for event in client.mark.delegate(resource, HighlightingJobParams(motivation="highlighting")):  # a job, followed
        match event:
            case JobCreated(data=created):
                print("to cancel it:", created.job_id)  # what client.job.cancel names
            case JobProgressed(data=progress):
                print(progress.percentage)
            case JobAttemptFailed(data=setback):
                print("trying again after:", setback.error)
            case JobCompleted(data=done):
                print(done.result)

    reached = await client.beckon.attention(resource, annotation)  # a drive: how many the gateway reached
    client.browse.click(annotation)  # a signal: this viewer's own, never sent
    print(reached)


async def over_http(origin: str, token: str, resource: ResourceId, annotation: AnnotationId) -> None:
    async with (
        HttpTransport(origin, token=Variable[str | None](token)) as transport,
        SemiontClient(transport, transport.content, transport) as client,
    ):
        await annotate(client, resource, annotation)

Every method returns one of nine shapes, and the table says which:

• A delegated job (mark.delegate, yield_.delegate) is described by its parameters, which are the protocol's own shapes: one for each motivation of a mark job (HighlightingJobParams, CommentingJobParams, AssessingJobParams, LinkingJobParams, TaggingJobParams), and GenerationJobParams for a yield job. Each takes what its job takes and nothing else. Awaited, the job gives its completion, which is its verb's: a MarkJobCompleteCommand, whose result is the job's counts (JobDetectionResult) or a decline (JobDeclinedResult), or a YieldJobCompleteCommand, whose result is the resource the job made (JobGenerationResult) or a decline. A type checker holds what reads a mark job's result to those two, and never offers it a generation's. Read, the job reports its progress and ends with that same completion. A job that says nothing is asked for its status, so a completion the stream did not carry is still heard. It ends as a JobError when the job failed for good, was cancelled, or (a generation) said nothing for longer than its length allows; and as a TransportError when what completed is not a job of the verb that was delegated.
• client.bus is the client's own bus: every frame its transport delivered, and every signal its own parts gave each other. client.wire is the bus over the transport, typed by channel.
• A client is held with async with: inside, it listens for what keeps its queries true, and it ends what it started on the way out. It does not close its transport: whoever opened that closes it, after the client. A session (session_from_kept) holds one as session.client.
• A client is generic in its transport. Code that takes a SemiontClient[Transport] runs over HTTP and over a test's doubles alike.
• What it sends unasked is the table's: a list's first hundred, a search's ten candidates, a context's two thousand characters. An option given as None is an option not given, and is not sent.

Why the shapes are these eight, in every SDK, is the reactive model.

Live queries

A query is one of ten reads (specs/src/client/refresh.json names them) that answer from the client's cache. Read once, it asks the knowledge base now. Held with async with, it is watched: its state now, and each state after it, until the client closes.

from typing import assert_never

from semiont.cache import Failed, Pending, Ready
from semiont.client import SemiontClient
from semiont.identifiers import ResourceId
from semiont.transport import Transport


async def watch(client: SemiontClient[Transport], resource: ResourceId) -> None:
    async with client.browse.annotations(resource) as live:  # holds the resource's scope
        async for state in live:
            match state:
                case Pending():
                    print("asking")
                case Ready(value=annotations):
                    print(len(annotations), "annotations")
                case Failed(error=error):  # a state, not a raise: the query lives on
                    print("failed:", error.code)
                case _:
                    assert_never(state)

• A state is one of three, and a match that leaves one out does not type-check. Failed is the state of a query with no value whose request failed and failed again. It is not the end: the next watcher, an event or query.invalidate() asks again, and every watcher is shown what comes.
• Watching is what asks. The first watcher of a query starts its request, and watchers that arrive together share it. A later one is given what is held. A request that fails is made once more; a query that has a value keeps showing it whatever its refetches do.
• A watched query stays true. While a query of one resource is held, the client holds that resource's scope, so its events reach the client; and each event asks again for exactly the queries the refresh table gives it, writes the value an event carries, or ends an annotation that is gone. What events ask of one query inside a second is one request. A stream that dropped and returned asks again for what its events could not replay.
• A read (await query.fresh()) asks now, raises the failure it meets, and gives every watcher of the query its answer.
• A watcher that falls behind is given the latest state, not each one it missed: a state is what is true now.
• A query is watched inside its client's async with, which is what listens for the events. After the client closes, a watcher is given nothing and a read is refused as bus.closed.
• A client can keep what its small queries hold (SemiontClient(..., persistence=CachePersistence(storage=..., key_prefix=...))), in a semiont.storage.SessionStorage the application supplies: a resource's description, its annotations, one annotation, the entity types, the tag schemas. The next client over the same storage shows them at once and asks for each anew the first time it is watched. session_from_kept(..., storage=...) does this for a session, and keeps the stream's place in each resource's scope with it (semiont.resume), so the next session is sent what was recorded since.

The behaviours every SDK's live queries are held to are numbered in the cache's semantics.

A worker

A worker claims jobs from a knowledge base's queue, does each one, and says what became of it. client.job.claim is its side of the queue. It is given what the worker accepts, each a JobFilter: the mark jobs of one motivation, or the yield jobs. What it returns hands out the jobs the worker comes to hold, one at a time, and a held job says its own lifecycle.

from semiont.annotations import QuotedText, annotation_of_span, reconcile
from semiont.claims import JOB_CLAIM_CHANNELS, JOB_COMMIT_CHANNELS, ClaimRefusal, HeldMarkJob, HeldYieldJob
from semiont.client import SemiontClient
from semiont.http import AgentToken, Credential, HttpTransport, ServiceToken
from semiont.identifiers import AnnotationId
from semiont.identity import agent_name
from semiont.types import Agent, AgentSoftware, Annotation, JobDetectionResult, JobProgress, MarkJobFilter, MarkJobFilterParams

PROVIDER, MODEL = "ollama", "gemma3:4b"


async def passages(text: str) -> list[QuotedText]:
    """Your model: the passages of a text it would highlight, each as the words it quoted. This one quotes the first line."""
    return [QuotedText(exact=text.partition("\n")[0])]


async def highlight(client: SemiontClient[HttpTransport], job: HeldMarkJob, generator: Agent) -> JobDetectionResult:
    """Your work: read the resource, have its passages quoted, and commit a highlight of each one the text has."""
    text = await client.browse.resource_content(job.resource_id)
    quoted = await passages(text)
    await job.progress(JobProgress(percentage=50))
    highlights: dict[AnnotationId, Annotation] = {}
    for quote in quoted:
        # What a model quotes is not trusted: it is found in the text, as the text has it, or it is dropped.
        span = reconcile(text, quote)
        if span is not None:
            built = annotation_of_span(text, span, resource_id=job.resource_id, motivation="highlighting", generator=generator)
            # An annotation's id is worked out from what it is, so a passage quoted twice is one annotation.
            highlights[built.id] = built
    await job.commit(job.resource_id, list(highlights.values()))
    return JobDetectionResult(found=len(quoted), persisted=len(highlights))


async def work(gateway: str, issuer: str, client_id: str, secret: str) -> None:
    # The process proves who it is at the issuer, and is given the agent its work is attributed to.
    service = ServiceToken(Credential(issuer=issuer, client_id=client_id, client_secret=secret))
    accepts = [MarkJobFilter(job_type="mark", params=MarkJobFilterParams(motivation="highlighting"))]
    async with (
        AgentToken(gateway, provider=PROVIDER, model=MODEL, service=service) as agent,
        # Its stream names what claiming and committing read. This worker awaits nothing else, so it names nothing else.
        HttpTransport(
            gateway, token=agent.token, refresher=agent.refresh, channels=(*JOB_CLAIM_CHANNELS, *JOB_COMMIT_CHANNELS)
        ) as transport,
        SemiontClient(transport, transport.content, transport) as client,
        # Leaving this block stops the worker: a job it still holds is failed first, and the queue retries it.
        client.job.claim(accepts) as claims,
    ):
        print(f"working as {agent.did}")
        # What made the annotations it commits: the agent the gateway gave it.
        generator = AgentSoftware(type="Software", id=agent.did, name=agent_name(PROVIDER, MODEL), provider=PROVIDER, model=MODEL)
        # Each job the worker comes to hold, one at a time. The next is claimed when this one settles.
        async for handed in claims:
            match handed:
                case ClaimRefusal(code="bus.unauthorized"):
                    # This credential can never claim. Stop, so that whoever runs the worker sees it.
                    raise PermissionError(handed.message)
                case ClaimRefusal():
                    print(f"claim refused: {handed.message}")
                case HeldYieldJob():
                    await handed.fail("this worker runs no yield job", failure_class="deterministic")
                case HeldMarkJob():
                    # A job left unsettled at the end of this block is failed.
                    async with handed as job:
                        await job.start()
                        try:
                            result = await highlight(client, job, generator)
                        except Exception as error:
                            await job.fail(str(error))
                        else:
                            await job.complete(result)

• A worker is an agent. Its process signs in with a service account, and the gateway gives it the agent its work is attributed to. The account needs two roles at the issuer: semiont-service, to be given an agent, and semiont-worker, without which every claim is refused as bus.unauthorized.
• Its stream names JOB_CLAIM_CHANNELS, and the reply channels of whatever else it awaits. The announcements that wake an idle worker reach only a stream that names them, so a client whose stream does not is refused at its first claim, as bus.unsubscribed.
• It builds what it commits with reconcile and annotation_of_span: Building annotations says what each does. The generator an annotation states is the agent the worker works as, whose DID is what the gateway says its token is.
• A held job commits for itself: await job.commit(resource_id, annotations) sends the batch as mark:commit, citing the job, and returns once the record has it. When no acknowledgement arrives it asks whether the batch's last annotation is on the resource, and a commit that is not established raises the failure of its unanswered request. The job says how its commits were established when it settles. A worker that commits names JOB_COMMIT_CHANNELS in its stream as well.
• It claims when it is idle: when its stream opens, each time a job settles, when a job it accepts is announced, and when its stream opens again. A claim answered with nothing pending is no refusal, and the worker waits. A ClaimRefusal is a claim the dispatcher refused or did not answer.
• A held job settles once: complete, with what its verb reports, fail, or cancel. Each says the outcome and lets the job go, and a second is refused. start comes first, and progress and checkpoint as often as there is something to say.
• A job is never left. async with job fails a job its block left unsettled, and leaving the claims' block stops the worker, which fails the job it still holds. The queue then runs it again at once, where a job whose worker was killed waits for the dispatcher's sweep.
• fail says whether the queue will run the job again, from the budget on the record the worker claimed and the failure's class: failure_class="deterministic" is a failure no second attempt can change, and failure_class="withheld" an answer the model's provider withheld. The queue retries neither.
• A cancellation is signalled. job.cancelled becomes true when a cancellation names the held job: the work stops where it can, and says await job.cancel().
• Each job has a trace of its own, and a worker's author writes nothing for it. A claim is made in no trace, whatever span the job before it was settled in, and async with job runs its block in the trace the job's reply arrived in, which is its claim's: a span a worker that exports telemetry starts in the block, and what the job sends from inside it, continue that trace. job.trace is the same trace as a value, a TraceContext (traceparent, and tracestate when there is one) or None for a reply that arrived in none: for what is done for the job where its block does not reach, such as another task, which enters it with with semiont.telemetry.continuing(job.trace):.
• claims.vitals() is what the worker can say of itself: when it last heard an announcement, claimed, was active and settled, the job it holds, and how many it has completed. claims.stalled tells of a held job that has shown no activity for fifteen minutes.

What a worker promises the dispatcher is the worker contract, and its conformance suite holds this package to it.

Building annotations

What a worker commits is annotations, and semiont.annotations builds them. It has three builders, the same three in every Semiont SDK, so that a highlight recorded by a worker in Python is the annotation a worker in TypeScript or in Rust records, with the same id.

from semiont.annotations import (
    QuotedText,
    annotation_exact_text,
    annotation_of_resource,
    annotation_of_span,
    body_source,
    is_highlight,
    reconcile,
    target_source,
)
from semiont.identifiers import ResourceId
from semiont.types import Agent, AnchoredText, Annotation, SpecificResource


def highlight(text: str, quoted: QuotedText, resource: ResourceId, generator: Agent) -> Annotation | None:
    """A highlight of the words a model quoted, or nothing when the text does not have them."""
    span = reconcile(text, quoted)  # where the words are, as the text has them, with the text's own context
    if span is None:
        return None
    return annotation_of_span(text, span, resource_id=resource, motivation="highlighting", generator=generator)


def highlight_of_a_pdf(anchored: AnchoredText, quoted: QuotedText, resource: ResourceId, generator: Agent) -> Annotation | None:
    """The same of a PDF: the words are found in its anchored text, and selected by rectangles on its pages."""
    span = reconcile(anchored.text, quoted)
    if span is None:
        return None
    return annotation_of_span(anchored, span, resource_id=resource, motivation="highlighting", generator=generator)


def link_to_what_was_generated(source: ResourceId, generated: ResourceId, generator: Agent) -> Annotation:
    """The link from a resource to one generated from it: an annotation of the resource as a whole."""
    names_it = SpecificResource(type="SpecificResource", source=generated, purpose="linking")
    return annotation_of_resource(source, motivation="linking", generator=generator, body=names_it)


def describe(annotation: Annotation) -> str:
    """What an annotation says, read without narrowing its target, its selector or its body by hand."""
    on = target_source(annotation.target)
    if is_highlight(annotation):
        return f"{annotation.id}: a highlight of {annotation_exact_text(annotation)!r} in {on}"
    return f"{annotation.id}: {annotation.motivation}, from {on} to {body_source(annotation.body)}"

• Nothing a model says is trusted. reconcile looks for the quoted words as they are. When the text does not have them so, it looks without regard to white space and to the form of quotation marks and dashes, then without regard to letter case, then for the nearest stretch within one edit for each twenty characters of the quote. Where the text has the words in more than one place, the model's prefix and suffix choose among them.
• A span is the text's own. A ReconciledSpan has where the words are (start, end), the words as the text has them (exact), what the text has on either side (prefix, suffix), and how it was found (anchor_method, and match_quality for a looser search). It is a TextSpan, and goes to annotation_of_span as it is.
• reconcile is one way to find a model's words, with its own tolerance for a misquote. A worker that finds its spans another way gives annotation_of_span a TextSpan it made itself.
• The builders make nothing a hand-written annotation could not, so nothing a knowledge base relies on rests on them.
• An offset counts Unicode code points, which is what a str is indexed by: text[span.start:span.end] is span.exact.
• The text it is counted in is client.browse.resource_content's: the resource's bytes read as a browser reads them, by the rule every Semiont reader keeps. The charset is the one the content type states and a label of the Encoding Standard, a byte-order mark decides over it, and what the encoding does not have is U+FFFD. A charset that is not decoded raises TransportError, and client.browse.resource_representation gives the bytes. A media type whose text is not its bytes raises it too: a PDF's text is its anchored text, and an image has none. semiont.resources.decode_representation(data, media_type) is the same reading, of bytes a program already holds.
• A span is checked before anything is built. One that is not the text's raises SpanRefusedError, and its code says which of seven refusals it is: span-out-of-range, exact-mismatch, prefix-mismatch, suffix-mismatch, and for a PDF nothing-located, exact-not-covered and item-out-of-range. The last is said of the PDF's anchored text and not of the span: one of its items is no stretch of its text, so no span of it is built. A span reconcile found in a text is not refused for that text.
• A PDF is annotated through its anchored text, which client.browse.resource_anchored_text answers. The words are found in its text, and the annotation selects them by a rectangle for each line they touch, and by their quote.
• An annotation's id is worked out from what it is: its resource, its motivation, its span and its body. Building the same annotation again gives the same id, and committing it again changes nothing, so a job that is run a second time writes no annotation twice.
• The readers read any annotation, built here or answered by a knowledge base, whatever shape its target, its selector and its body take: target_source, annotation_exact_text, body_source, is_highlight and the rest.

The rules are case tables that every SDK runs: reconcile-cases.json, builder-cases.json and id-cases.json.

Testing what is built on it

semiont.testing gives a real client over doubles. Its namespaces, its cache and its deadlines are the client's own; only what it speaks through is scripted. A test says what the knowledge base answers, runs the code under test, and reads what that code asked.

from semiont.client import SemiontClient
from semiont.identifiers import ResourceId
from semiont.testing import create_test_client
from semiont.transport import Transport


async def title_of(client: SemiontClient[Transport], resource: ResourceId) -> str:
    return (await client.browse.resource(resource).fresh()).name.title()


async def a_title_is_its_resources_name_in_title_case() -> None:
    made = create_test_client()
    made.transport.queue_reply(
        "browse:resource-requested",
        [
            {
                "resource": {"@context": "https://schema.org", "@id": "res-1", "name": "a page of notes", "representations": []},
                "annotations": [],
                "entityReferences": [],
            }
        ],
    )

    async with made.client as client:
        assert await title_of(client, ResourceId("res-1")) == "A Page Of Notes"

    # What the code under test asked of the knowledge base, as it was sent.
    assert [(asked.channel, asked.payload) for asked in made.transport.request_log] == [
        ("browse:resource-requested", {"resourceId": "res-1"})
    ]
    await made.transport.close()

• FaultyTransport is a transport with no wire. queue_reply says what the gateway answers the next requests of an operation with, and refuse_when has it answer with a failure. A reply that names what it answers for, as a gathered context names its resource, takes that from the request, so a test queues the response alone. A schedule of Deliver, DropReply, Delay, DuplicateReply and RejectEmit says what the wire does to each request in turn, so a lost reply and the retry after it can be scripted. request_log and emitted are what was sent; deliver carries a frame in as the bus would; set_state drops the stream and brings it back.
• InMemoryContent keeps what is uploaded and what a test seeds, and StubGateway answers the gateway's own operations as a test scripted them. Each records the calls made of it.
• A double refuses what nobody scripted, by name. A request with no answer scripted fails naming its operation, and a read of content nobody stored fails naming the resource. None answers with a value of its own making, so a test that forgot to say something fails where it forgot.
• create_test_client takes the doubles a test built, or makes ones nothing is scripted on, and the client's own timing and persistence.

The bus

Under the namespaces is the bus, and a program can speak on it directly. A transport is held with async with. Its stream opens inside, once there is a token, and everything it started has ended on the way out.

from semiont.bus import Bus
from semiont.channels import MARK_ADDED
from semiont.errors import BusRequestError
from semiont.http import HttpTransport
from semiont.identifiers import ResourceId
from semiont.operations import BROWSE_RESOURCE_REQUESTED
from semiont.types import BrowseResourceRequest
from semiont.watched import Variable


async def read(origin: str, token: str, resource: ResourceId) -> None:
    async with HttpTransport(origin, token=Variable[str | None](token)) as transport:
        bus = Bus(transport)
        try:
            reply = await bus.request(BROWSE_RESOURCE_REQUESTED, BrowseResourceRequest(resource_id=resource))
        except BusRequestError as error:
            print(error.code)  # "bus.not-found", "bus.timeout", …: a closed vocabulary
            return
        print(reply.response.resource.name)

        # A resource's channel is read for the resource: the read holds its scope, and lets go on the way out.
        async with bus.frames(MARK_ADDED, resource) as added:
            async for frame in added:
                print(frame.scope, frame.payload.type)

• The bus is typed by channel. A channel is a constant that carries its payload's type: a request is refused another operation's payload, its reply is that operation's result, and a frame's payload is its channel's. By name, with JSON objects, it is semiont.bus.request and the transport's own emit and frames.
• A request waits for the stream to be open before it is sent, is answered once, and is abandoned by cancelling the task that awaits it.
• Frames are a sequence: each reader is given every one, in order, however far behind it falls.
• A resource's channel is read for the resource. mark:added and the other events of the record are delivered on their resource's scope, and their constants are ScopedChannels. bus.frames(MARK_ADDED, resource) holds that scope until the read ends, and gives that resource's frames and no other's. With no resource it does not type-check: such a read would be given nothing, and say nothing of it.
• A recorded event's own payload is its channel's to give. A frame of mark:added carries the stored event, which holds the event's own payload as JSON. MARK_ADDED.event(stored) is that payload as an AnnotationAddedPayload, for the stored event of a frame or of a history (browse:events-requested). A stored event of another type is refused.
• The connection's state (transport.state) is a value: what it is now, and each thing it becomes. A dropped stream is a state, never an error, and opens again by itself.
• Failures are SemiontErrors under the codes of specs/src/errors/codes.json. What a server refused is also on transport.failures().
• The token is a value the transport watches: when it changes, the next request carries the new one. A stream the gateway refused as unauthorized asks its refresher for another token, once per outage, and with none waits, with no request, for a different one.

The protocol it keeps to is the transport contract and the HTTP transport, and the conformance suite holds it there, case by case, beside the TypeScript and Rust SDKs.

Content, and the gateway's own operations

Bytes never ride the bus. A transport's content uploads them and reads them back, and the transport itself asks the gateway what a gateway answers for itself.

from semiont.http import HttpTransport
from semiont.identifiers import ResourceId
from semiont.transport import PutBinaryRequest


async def store(transport: HttpTransport, page: bytes) -> ResourceId:
    upload = transport.content.put_binary(
        PutBinaryRequest(name="A page", file=page, format="text/markdown", storage_uri="file://pages/a.md", entity_types=["Note"])
    )
    async for progress in upload:
        print(progress.bytes_uploaded, "of", progress.total_bytes)
    created = await upload

    stored = await transport.content.get_binary(created.resource_id)
    assert (stored.data, stored.content_type) == (page, "text/markdown")
    async with await transport.content.get_binary_stream(created.resource_id) as arriving:
        async for piece in arriving:
            print(len(piece))

    me = await transport.get_current_user()
    print(me.did, (await transport.health_check()).status)
    return created.resource_id

• An upload is awaited for the resource it created and iterated for its progress, each once. Its caller abandons it by cancelling the task that does either: its connection closes, and nothing is reported.
• What is read is what was stored, byte for byte, with its media type. A time or a URI in an answer is the text the gateway sent, and a field it sent as null is None.
• A resource's description and its anchored text are read here too, by get_resource_graph and get_anchored_text, which are what client.browse.resource_graph and client.browse.resource_anchored_text call. An anchored text is answered as the text with where each run of it is on a page, as a stored decline, or as why there is none: an absence is an answer, and no failure.
• A read is asked a second time when the gateway says it will recover, or says nothing. A request that may already have had its effect is not.

Ids, and the protocol's shapes

The package holds what the protocol states, as Python types: the kinds of id, the shapes the protocol sends and answers with, the channels of its bus and the requests made over them. Each kind of id is a type of its own, made only from text that is one, and a type checker refuses one kind where a shape names another.

from semiont.identifiers import AnnotationId, InvalidIdentifier, ResourceId
from semiont.types import MarkDeleteCommand

resource = ResourceId("5bcd259ab1464cf68a556bbad21f513f")
assert ResourceId.parse("not an id") is None  # where text that is not an id is an ordinary answer
try:
    ResourceId("https://kb.example/resources/x")
except InvalidIdentifier as refused:
    print(refused)  # 'https://kb.example/resources/x' is not a ResourceId: it does not match ^[A-Za-z0-9_-]{1,128}$

command = MarkDeleteCommand(annotation_id=AnnotationId("a-1"), resource_id=resource)
print(command.model_dump(mode="json", exclude_unset=True))
# {'annotationId': 'a-1', 'resourceId': '5bcd259ab1464cf68a556bbad21f513f'}

# MarkDeleteCommand(annotation_id=resource) is refused by a type checker: a ResourceId is not an AnnotationId.

Telemetry

The SDK takes OpenTelemetry's API and installs nothing: it does nothing until the application it runs in installs a provider. With one, it exports what specs/src/sdk-telemetry/telemetry.json lists, as every Semiont SDK does: a span for each frame sent and each received, a span for each upload and each read, and a count of the emits sent. A request carries the trace it is made in, and a frame continues the trace it was sent under (frame.trace). A job a worker holds states the trace it is run in (job.trace), and runs the block that holds it in that trace. semiont.telemetry.continuing(trace) holds either for as long as its with block lasts.

With SEMIONT_BUS_LOG set to any value that is not empty, a transport over HTTP also writes the bus log, as every Semiont SDK does: one line on stdout for each frame it sends and each it receives, and for each read and each upload of content. A line says the channel, the resource's scope, the first eight characters of the correlation id and of the trace's id, and the payload as JSON:

[bus EMIT] job:claim cid=a89a670a {"accepts":[{"jobType":"yield"}]}
[bus RECV] job:claim-failed cid=a89a670a {"code":"none-pending","message":"No pending job matches"}

The variable is read once, when the first line would be written. No line holds a credential: the value of a payload's member whose name is one a credential goes by (token, which is what a payload names the token a resource is cloned by, password, apiKey and the rest of the short list every bus log keeps) is written as [withheld].

What is in the package

Nothing here restates the protocol by hand. identifiers, types, channels, operations, error_codes, timing, refresh, telemetry_table, oauth_clients, sign_in, media_types_table and _encodings (the Encoding Standard's labels and single-byte encodings, by which a resource's text is read) are generated from the specification, and a gate in CI fails when one of them differs from what the specification generates.

Working on it

The package is packages/sdk-python of the Semiont repository, and Testing says how its checks and its tests are run. The programs this page shows are files of its tests, word for word: both type checkers check them, and each one is run.

License

Apache-2.0. See LICENSE.