> Canonical guide: https://developers.foxlight.ai/build/tutorials/video-studio/workshop/ > Contract snapshots: Skulk 2.0.0 (b0af39c79b6b7102b2478062904f1d7cc8619975); SDK 0.4.0 (31bb090b8f64689f87514e48385e22b6ad94a6c2). Check the installed runtime when versions differ. # 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](https://developers.foxlight.ai/build/get-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: ```bash python workshop.py ``` Expected output: ```text PASS: one submission, validated result, authority and replay checks ``` ## Complete program ```python title="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](https://developers.foxlight.ai/build/tutorials/video-studio/testing/).