Targets and Interceptors#

A capture target is a directory under the server root containing a __capture_target__.py file. The directory's path relative to the root is the target's name, so directories double as groups: Plants/Tulip and Plants/Sunflower appear together under Plants in the app and the panel. Everything else about a target — captured sessions, interceptor state, history — lives in the same directory:

my-deployment/
├── config.toml
└── PlantJournal/
    ├── __capture_target__.py     ← the target definition (you write this)
    ├── .persistence.db           ← interceptor state (created when the target loads)
    ├── .history/                 ← snapshots of validated saves
    └── sessions/                 ← schema-defined directories of captured files
        ├── 2026-06-14/
        │   ├── photo.jpeg
        │   └── notes.json
        └── ...

A target has no separate manifest file. Its on-disk layout is its schema — the cg.Struct you declare — and the schema travels with the procedure definition the app downloads.

The main procedure#

__capture_target__.py declares a schema, records a procedure against it with @cg.procedure, and marks that procedure as the target's main procedure with @cgserver.target. There is exactly one per module:

"""A target that captures a daily photo and a short note."""

import capturegraph as cg
from server import authoring as cgserver


class Session(cg.Struct):
    photo: cg.Image
    notes: cg.String


class PlantJournal(cg.Struct):
    sessions: cg.Map[cg.Date, Session]
    _saw_instructions: cg.Bool
    _metadata: cg.Metadata


@cgserver.target
@cg.procedure(PlantJournal)
def plant_journal(root):
    root._saw_instructions |= cg.ShowInstructions(
        text="Photograph the plant from the marked spot."
    )
    session = root.sessions[cg.CaptureTime()]
    session.photo &= cg.CaptureImage(label="Photograph the plant")
    session.notes &= cg.UserInputString(label="Notes")

@cgserver.target marks the recorded procedure as the target's main procedure and returns it unchanged. The server reads its {schema, root, nodes} wire form straight off the function and serves it at GET /api/v1/targets/{target}/definition — the schema is embedded in that response, so it doubles as the data layout the app, panel, and sync tool walk.

Drop this directory into the server root and it is hosted within target_refresh seconds — no restart. The server re-executes this module as soon as the file changes on disk (it stats the file per request and reloads on a change), so saving an edit is deploying it: a participant pulling to refresh sees the new procedure within seconds.

Recording against the schema#

Inside the body, root is a Path into the schema. You navigate it by field, index, and key, and cache results into positions with two operators:

• path &= value — overwrite: capture this fresh in every session.
• path |= value — skip if it already exists: configuration, references, and once-per-target values.

cg.do(step) records a side-effect-only step (a metronome, an assertion). Instructions are the exception: cg.ShowInstructions returns the user's acknowledgement, so cache it with |= into a hidden cg.Bool field (as above) to show it once instead of every session. Control flow uses context managers: with cg.when(cond):, with cg.optional():, and with cg.while_repeat(cond, array_path) as elem:. See the DSL guide for the full recorder surface.

Interceptors#

An interceptor fills one schema position dynamically: instead of reading that position's file from disk, the server computes it on every download. This is the hook that turns a passive file host into a capture coordinator — the _metadata._next_notification slot above is a cg.Time, and an interceptor decides, per user and per request, what that time should be.

Declare one with @cgserver.intercept(Schema, selector). You name the position with a typed selector that navigates the target schema from its root; the selector's return type is the value the function fills, so the slot and its type are fixed statically — you never pass the type:

import capturegraph as cg
from server import authoring as cgserver


@cgserver.intercept(PlantJournal, lambda s: s._metadata._next_notification)
def notification(profile) -> cg.Time:
    # the _next_notification slot is a cg.Time, so wrap the datetime
    return cg.Time(next_capture_time)

The selector runs once at import against a tracer, so a field the schema does not have fails the target load there — never as a silent missing file at serve time. Selecting the schema root (lambda s: s) or navigating into a non-struct field is likewise rejected at decoration.

The positions the iOS app polls automatically are s._metadata._next_notification (a cg.Time, the next capture time) and s._metadata._location (a cg.Location, the next capture point).

Hidden (_-prefixed) schema fields map to dotfiles on disk, so _metadata._next_notification is served as .metadata/.next_notification.

Function parameters#

An interceptor may declare any subset of four parameter names, in any order; the server passes exactly the ones the signature asks for, as keyword arguments:

So def notification(profile, state) and def count(target) are both valid. Declaring any other required parameter raises InterceptorDefinitionError at import time.

Persistence#

state is a Scope — a namespaced key/value store backed by the target's .persistence.db (SQLite in WAL mode). The whole model in four rules:

1. One transaction per call. Each interceptor call reads, modifies, and writes inside a single BEGIN IMMEDIATE transaction, so concurrent requests serialize across worker processes and a raise rolls the whole change back.
2. Private by construction. Each interceptor's state is its own child namespace (keyed by the slot it fills), so two interceptors can use the same key names without ever colliding. Declare shared instead for the one namespace every interceptor of the target sees.
3. JSON-native values only. Encode instants as ISO strings (or via the cg codecs) before storing them.
4. Regenerable bookkeeping, never an archive. Any consumer must be able to rebuild what it needs from an empty scope on the next request. That is what lets a consumer that changes the shape of what it stores discard old state instead of migrating it — the scheduling books stamp their scope with a format version (ensure_format) and reset it on a mismatch, and deleting .persistence.db outright is always safe (at the cost of forgetting reservations in flight).

seen = state.get("seen", 0)
state["seen"] = seen + 1
state["last_user"] = profile.user_id

Call state.scope("name") for a further child namespace when one interceptor keeps several independent consumers apart — e.g. hand a scheduling book a state.scope("schedule") of its own so its format resets never touch your other keys.

To see what a live target has persisted, read its database directly — each row is one key, with namespaces flattened into the scope column (an interceptor's private namespace appears as interceptor:<field path>, the shared one as shared):

sqlite3 path/to/target/.persistence.db "SELECT scope, key, value FROM state"

What the server enforces#

• Position by path. A request is intercepted only when its full relative path matches the declared position — a file with the same name elsewhere is served from disk as usual.
• Session paths are never intercepted. Files inside capture sessions always go to disk.
• Type resolution + validation. The result type is resolved from the schema at the field path, and the value is coerced to it before serving. A bare value (a datetime for a cg.Time slot, a number for cg.Number) is wrapped automatically.
• Write protection. A client upload to an intercepted path is refused with HTTP 409.
• Delivered dynamically. On the wire an intercepted position resolves as a dynamic leaf: a present verdict with dynamic: true, its freshly computed version, extension "json", and bytes always carried inline — the server recomputes it per request, so a client must not poll it for change in the background. See the captured-data format for the full contract.

Example: scheduling notifications#

Interceptors are where the scheduling library runs. This notification interceptor spreads captures across solar conditions with select_sessions (Void & Cluster) and gives each participant their own time with Reservations:

from datetime import datetime, timedelta

import capturegraph as cg
import capturegraph.scheduling as cgsh
from server import authoring as cgserver


@cgserver.intercept(PlantJournal, lambda s: s._metadata._next_notification)
def notification(profile, target, state) -> cg.Time:
    location = target._metadata._location

    # Sessions already taken, described by their solar conditions.
    previous = cg.Array(
        [
            {"date": when, "solar_angle": cgsh.forecast.solar_position(location, when)}
            for when in target.sessions.keys()
        ]
    )

    # Candidate times over the next 24 hours, with solar angles.
    potential = cg.Array(
        [
            {"date": when, "solar_angle": cgsh.forecast.solar_position(location, when)}
            for when in cgsh.forecast.times(span=timedelta(hours=24))
        ]
    )

    # Void & Cluster: prefer conditions unlike anything captured yet.
    selected = cgsh.select_sessions(
        potential_sessions=potential,
        previous_sessions=previous,
        distance_fn=cgsh.distance.combine(solar_angle=cgsh.distance.solar(sigma_deg=2.0)),
        selections=10,
    )

    # One distinct time per user, released if they don't capture in time.
    reservations = cgsh.Reservations(
        state.scope("schedule"),
        grace=timedelta(minutes=30),
    )
    chosen = reservations.reserve(profile.user_id, among=[s["date"] for s in selected])
    if chosen is None:
        return cg.Time(datetime.now() + timedelta(hours=1))
    return cg.Time(chosen)

target._metadata._location reads the target's capture-location position — which can itself be an interceptor: a GPS-survey target declares an @cgserver.intercept(Schema, lambda s: s._metadata._location) in the same module that assigns each user their next capture point with CoverageReservations. See Reservations for the full scheduling walkthrough.

Editing safely: the validation pipeline#

Procedure saves from the management panel never touch the live file directly:

1. The candidate is written to a hidden pending file at the target root.
2. A subprocess (validate_target.py) imports it, finds and builds the @cgserver.target procedure, serializes the result, and checks every interceptor declaration — killed after validation_timeout seconds if it hangs.
3. Only on success is the current live file snapshotted into .history/ (newest history_limit kept) and the pending file swapped in atomically.

A broken save therefore can never brick a live target, and every previous version is one click away in the panel's History view — restores run through the same pipeline, so they are themselves undoable.

Editing __capture_target__.py directly on disk (over SSH, say) skips all of this: the edit goes live on the next request, valid or not. That is sometimes what you want, but the panel's editor is the safer default.