9. Run a local operation exercise

Now build the smallest useful part of Studio: an operation service connected to a backend adapter. This complete example uses the public SDK interface. It creates no video and needs no Skulk™ cluster, GPU or model.

Prerequisites and run command

Obtain the SDK through Get the SDK and install its supplied wheel in a Python 3.13+ environment. Use the interpreter where that wheel is installed. This example is checked against SDK 0.4.0; an unrelated package with a similar name is not a substitute.

Copy the code below into workshop.py, then run:

python workshop.py

Expected output:

PASS: one submission, validated result, authority and replay checks

Complete program

workshop.py

import asyncio
import json
import tempfile
import time
from pathlib import Path

from pydantic import Field, JsonValue, ValidationError
from skulk_capability_sdk.contracts import Contract, OperationBounds
from skulk_capability_sdk.operations import (
    OperationState,
    OperationCall,
    OperationJournal,
    OperationLogs,
    OperationPlan,
    OperationRunner,
    OperationService,
    OperationStatus,
    SubmissionRefused,
    input_digest,
)
from skulk_capability_sdk.storage import secure_directory


Observation = tuple[
    OperationState, float | None, str | None, dict[str, JsonValue] | None, str | None
]


class Shot(Contract):
    """Input for our simulated render."""

    prompt: str = Field(min_length=1, max_length=200)


class Clip(Contract):
    """Metadata only: this exercise does not create a video file."""

    label: str = Field(min_length=1, max_length=200)


class FakeBackend:
    """A tiny backend whose jobs live only in this process."""

    def __init__(self) -> None:
        self.submissions = 0
        self.jobs: dict[str, tuple[str, int, bool]] = {}

    async def submit(self, operation_id: str, payload: dict[str, JsonValue]) -> str:
        try:
            shot = Shot.model_validate_json(json.dumps(payload))
        except ValidationError as error:
            raise SubmissionRefused("invalid shot") from error
        self.submissions += 1
        self.jobs[operation_id] = (shot.prompt, 0, False)
        return operation_id

    async def observe(self, reference: str) -> Observation:
        label, ticks, cancelled = self.jobs[reference]
        if cancelled:
            return "cancelled", None, "stopped", None, None
        if ticks >= 3:
            return "succeeded", 1.0, "done", {"label": label}, None
        self.jobs[reference] = (label, ticks + 1, False)
        return "running", ticks / 3, "simulating", None, None

    async def cancel(self, reference: str) -> None:
        label, ticks, _ = self.jobs[reference]
        self.jobs[reference] = (label, ticks, True)


async def demo(root: Path) -> None:
    """Prove reservation, authority and result handling without inference."""
    secure_directory(root)
    scope = "example-install/example.render@1.0.0"
    bounds = OperationBounds(
        max_queued=2, max_runtime_seconds=30,
        stale_after_seconds=5, retain=4, log_bytes=1024,
    )
    backend = FakeBackend()
    journal = OperationJournal(root / "operations.sqlite", bounds)
    logs = OperationLogs(root / "logs", bounds)
    runner = OperationRunner(
        journal, logs, backend, scope=scope,
        poll_seconds=0.02, adapter_seconds=1,
        result_schema=Clip.model_json_schema(),
    )

    async def plan(payload: dict[str, JsonValue]) -> OperationPlan:
        shot = Shot.model_validate_json(json.dumps(payload))
        return OperationPlan(
            plan_digest=input_digest(scope, payload),
            summary=shot.prompt, expires_at=int(time.time()) + 300,
            input=payload,
        )

    service = OperationService(
        scope=scope, journal=journal, logs=logs, runner=runner, plan=plan,
    )
    runner.start()
    try:
        payload: dict[str, JsonValue] = {"prompt": "A fox in a snowy forest"}
        reviewed = await service.dispatch(OperationCall(op="plan", input=payload), "effect")
        assert isinstance(reviewed, OperationPlan)
        call = OperationCall(op="start", operation_id="take-001", input=reviewed.input)
        await service.dispatch(call, "effect")
        await service.dispatch(call, "effect")  # Lost-reply retry: same id and input.

        async with asyncio.timeout(5):
            while True:
                status = await service.dispatch(
                    OperationCall(op="status", operation_id="take-001"), "read",
                )
                assert isinstance(status, OperationStatus)
                if status.state == "succeeded":
                    break
                assert status.state in ("queued", "running"), status
                await asyncio.sleep(0.02)
        assert backend.submissions == 1
        assert status.result == {"label": payload["prompt"]}

        try:
            await service.dispatch(call, "read")
        except PermissionError:
            pass
        else:
            raise AssertionError("read authority unexpectedly started work")

        try:
            await service.dispatch(OperationCall(
                op="start", operation_id="take-001", input={"prompt": "Different shot"},
            ), "effect")
        except ValueError:
            pass
        else:
            raise AssertionError("an operation id was reused with different input")
        print("PASS: one submission, validated result, authority and replay checks")
    finally:
        await runner.stop()


if __name__ == "__main__":
    with tempfile.TemporaryDirectory() as directory:
        asyncio.run(demo(Path(directory) / "studio-exercise"))

Read it from the outside inward

1. Shot and Clip define what this application accepts and produces. The result contains metadata, not fake MP4 bytes.
2. FakeBackend implements the adapter's three methods: submit, observe and cancel.
3. The journal persists intent and results; logs have their own bounded storage.
4. The runner owns background work and validates completion against Clip's schema.
5. The service exposes operation verbs, enforcing authority and reservation rules.
6. The caller reviews, starts, retries the same id and polls using read authority.

The service and runner must share the same scope. A service must never enqueue into a worker configured for another contract. The timeout makes this exercise fail visibly if the runner stops progressing.

The fake planner does not select a model or enforce a reviewed-plan digest at submission. It validates the input and demonstrates the common plan envelope. Studio's real adapter adds live resolution and digest comparison.

This is a local exercise, so it calls service.dispatch directly with explicit authority. In a managed application, authority must follow the host's admission contract. Studio's current governed host admits calls before child dispatch, and Studio passes "effect" to its services. That is a property of that host contract, not permission for an arbitrary HTTP caller to promote itself.

What the checks prove

• A repeated start with the same id and input leads to one backend submission.
• A completed result matches the declared output schema.
• Read authority cannot start work.
• Reusing an operation id for different input is refused.

The journal is durable, but the fake backend's dictionary is not. This program does not qualify restart recovery, real media collection or uncertain network submission. A production adapter needs an independently observable backend reference for those cases.

Three extensions to try

Cancellation: Start another id and cancel while it is running. Poll until the backend's observation establishes the terminal outcome. A cancel request alone is not proof that work stopped.

Refusal: Submit an empty prompt. Confirm the adapter refuses before incrementing submissions, and the operation reports failure.

Bad output: Return an object without label from the fake backend. Confirm result-schema validation prevents a succeeded operation with an invalid result.

After these work, replace FakeBackend with an HTTP job adapter. Follow Studio's distinction between a definite pre-send refusal, an accepted reference and an uncertain submission.

Next: testing and diagnosis.