Metadata-Version: 2.4
Name: osp-provider-sdk
Version: 0.1.0
Summary: Optional declaration, planning and progress helpers for OSP providers.
Author: OSP Team
License: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.13
Requires-Dist: loguru<1,>=0.7
Requires-Dist: osp-provider-contracts<0.10,>=0.9.0
Requires-Dist: osp-provider-runtime<0.11,>=0.10.0
Requires-Dist: pydantic<3,>=2.11.7
Requires-Dist: pyyaml<7,>=6.0.2
Provides-Extra: dev
Requires-Dist: build<2,>=1.2; extra == 'dev'
Requires-Dist: hatch<2,>=1.14; extra == 'dev'
Requires-Dist: pytest<10,>=9.0.3; extra == 'dev'
Requires-Dist: ruff>=0.15; extra == 'dev'
Requires-Dist: twine<7,>=6; extra == 'dev'
Requires-Dist: ty>=0.0.89; extra == 'dev'
Requires-Dist: uv<0.13,>=0.12.5; extra == 'dev'
Description-Content-Type: text/markdown

# osp-provider-sdk

A thin, optional authoring layer for OSP providers. Declare actions and lists,
plan before effects, and report progress through the existing Provider protocol.
The package layers are **contracts → runtime → SDK → providers**. The
orchestrator never installs the SDK; existing providers can keep their own
Provider implementations.

```bash
pip install osp-provider-sdk
```

## One-file provider

This is [examples/buckets.py](examples/buckets.py). Replace `backend` with your
team's adapter. The included [backend](examples/backend.py) is in-memory and
makes no network calls; real backends must persist the task's client token for
recovery across processes. Do not adopt a bucket merely because its name matches.

```python
"""S3-style bucket provider on the OSP SDK, with a replaceable backend adapter."""

from contextlib import asynccontextmanager
from pathlib import Path
from typing import Literal

from osp_provider_contracts.canonical_digest import canonical_digest
from pydantic import BaseModel, ConfigDict, Field

from osp_provider_sdk import App, Done, Page, Plan, Resource, ResourceOwnership, Violation

from . import backend


@asynccontextmanager
async def lifespan():
    async with backend.Client.from_env() as client:
        yield client


app = App("s3", policy=Path(__file__).with_name("policy.yaml"), lifespan=lifespan)


class Create(BaseModel):
    model_config = ConfigDict(frozen=True)
    quota_gib: int = Field(10, ge=1, le=1000, title="Quota (GiB)")
    region: Literal["osl", "bgo"] = "osl"


class Resize(BaseModel):
    model_config = ConfigDict(frozen=True)
    quota_gib: int = Field(ge=1, le=1000, title="Quota (GiB)")


class Row(BaseModel):
    quota_gib: int = Field(title="Quota (GiB)")
    region: str


def _row(b) -> Resource:
    return Resource(
        target_ref=b.id,
        display=b.name,
        ownership=ResourceOwnership(contacts=b.contacts),
        fields=Row(quota_gib=b.quota_gib, region=b.region).model_dump(),
        created_at=b.created,
    )


@app.resource("bucket", fields=Row, filters=["region"])
async def buckets(q, access) -> Page:
    page = await app.state.list(region=q.filters.get("region"), cursor=q.cursor, limit=q.limit)
    return Page(
        tuple(_row(b) for b in page.items if access.allows(b.contacts)),
        page.cursor,
        observed_at=page.observed_at,
    )


@buckets.get
async def bucket(ref, access) -> Resource | None:
    b = await app.state.get(ref)
    return _row(b) if b and access.allows(b.contacts) else None


@app.action("create", inputs=Create, stages=["s3.create", "s3.tag"])
async def create(req, ctx) -> Done:
    with ctx.step(
        "s3.create",
        f"Creating bucket {req.name}: {req.inputs.quota_gib} GiB in {req.inputs.region}",
    ):
        b = await app.state.create(
            req.name, req.inputs.region, req.inputs.quota_gib, client_token=ctx.task_ref
        )
        ctx.created(b.id, req.name)
    with ctx.step("s3.tag", f"Recording owners {', '.join(ctx.contacts)} on {req.name}"):
        await app.state.tag(b.id, contacts=ctx.contacts)
    return Done(f"Created bucket {req.name} ({b.id})")


@create.plan
async def plan_create(req, ctx) -> Plan:
    if not req.name:
        return Plan(
            "Missing bucket name",
            violations=(Violation.missing_required("name", "A bucket name is required"),),
        )
    taken = await app.state.get_by_name(req.name)
    if taken and taken.client_token != ctx.task_ref:
        return Plan(
            f"bucket {req.name}: name taken",
            violations=(
                Violation.invalid_value("name", f"Bucket {req.name} already exists ({taken.id})."),
            ),
        )
    return Plan(
        f'create bucket "{req.name}" in {req.inputs.region}, quota {req.inputs.quota_gib} GiB, '
        f"owners {', '.join(sorted(ctx.contacts))}"
    )


@app.action("resize", target="resource", inputs=Resize)
async def resize(req, ctx) -> Done:
    with ctx.step(
        "s3.quota", f"Setting {req.target_ref} quota {ctx.plan.data} → {req.inputs.quota_gib} GiB"
    ):
        await app.state.set_quota(req.target_ref, req.inputs.quota_gib)
    return Done(f"Bucket {req.target_ref} quota is {req.inputs.quota_gib} GiB")


@resize.plan
async def plan_resize(req, ctx) -> Plan:
    b = await app.state.get(req.target_ref)
    if b is None:
        return Plan(
            "Bucket not found",
            violations=(Violation.invalid_value("target_ref", "Bucket does not exist"),),
        )
    if b.quota_gib == req.inputs.quota_gib:
        return Plan.nothing_to_do(f'bucket {b.id} "{b.name}": quota already {b.quota_gib} GiB')
    return Plan(
        f'bucket {b.id} "{b.name}": quota {b.quota_gib} → {req.inputs.quota_gib} GiB',
        data=b.quota_gib,
    )


@app.action(
    "destroy",
    target="resource",
    destructive=True,
    cancellable=True,
    stages=["s3.empty", "s3.delete"],
)
async def destroy(req, ctx) -> Done:
    with ctx.step("s3.empty", f"Emptying bucket {req.target_ref}"):
        async for done, total in app.state.empty(req.target_ref, objects=ctx.plan.data):
            ctx.progress(done, total, f"Emptied {done} of {total} prefixes in {req.target_ref}")
            ctx.checkpoint()  # The previous prefix delete has settled.
    with ctx.step("s3.delete", f"Deleting bucket {req.target_ref}"):
        await app.state.delete(req.target_ref)
    return Done(f"Deleted bucket {req.target_ref}")


@destroy.plan
async def plan_destroy(req, ctx) -> Plan:
    b = await app.state.get(req.target_ref)
    if b is None:
        return Plan.nothing_to_do(f"bucket {req.target_ref}: already absent")
    objects = frozenset(b.objects)
    scope = canonical_digest(sorted(objects))
    return Plan(
        f'delete bucket {b.id} "{b.name}" and {len(objects)} object versions (scope {scope})',
        data=objects,
    )


if __name__ == "__main__":
    app.run()
```

The matching [policy.yaml](examples/policy.yaml) is loaded with `yaml.safe_load`
and validated against the generated manifest before entering lifespan:

```yaml
request_policy:
  version: 1
  rules:
    - id: team_access
      match: {kind: bucket, actions: [create, resize, destroy], instance: dev}
      check: {netgroup_in: [team-storage]}
      level: gate
      approval: {kind: maintainer, required_approvals: 1}
      message: This team needs a storage operator's approval.
    - id: quota_warning
      match: {kind: bucket, actions: [create, resize]}
      check: {input: quota_gib, max: 80}
      level: warn
      message: This is a large bucket quota.
    - id: quota_approval
      match: {kind: bucket, actions: [create, resize]}
      check: {input: quota_gib, max: 100}
      level: gate
      approval: {kind: maintainer, required_approvals: 1}
      message: Quotas above 100 GiB need storage operator approval.
```

The default destructive-action Gate requires one Maintainer approval. To change
it provider-wide, add `delete_approval: {kind: admin, required_approvals: 1}`
under `request_policy`. `peer` is also supported. Rules belong to contracts;
backend-specific decisions stay in the custom read-only planner.

## Invocation and lifetime

`app.run()` reads `RABBITMQ_URL`, `PROVIDER_INSTANCE`, `PROVIDER_INSTANCE_ID`
and `CONCURRENCY` (default 1). Queue names come from `ProviderIdentity`.
For example, `PROVIDER_INSTANCE=dev python -m examples.buckets` starts the
example's RabbitMQ runner when `RABBITMQ_URL` is provided.

Offline invocation uses `with app.running():`. This validates declarations,
enters lifespan, and owns one event loop if any callback or lifespan is async.
All async backend clients are created and closed on that loop. Runtime workers
block for their handler's result, so runtime concurrency still bounds work.
Async handlers must move blocking backend calls into `asyncio.to_thread`.
Sync handlers returning awaitables fail with an instruction to use `async def`.

```python
from osp_provider_contracts import ProviderRequest, RequestContext
from examples.buckets import app

with app.running():
    result = app.execute(
        ProviderRequest("create", name="logs"),
        RequestContext("local-task", netgroup="team-storage", contacts=("storage@example.org",)),
    )
    print(result.produced_resources)
```

## Planning and approval

Assessment, CHECK and APPLY use one planner: model validation → shared rules →
custom plan → combination. Assessment always describes APPLY intent and has no
cleared approvals. It returns structured rejection, incomplete or approval
facts without calling apply. CHECK and no-op plans stop before apply.

Gated actions must declare `.plan` and author a concrete summary. The Gate key
names a requirement; its fingerprint covers every frozen Gate field except the
key, including summary and effective inputs. Use native IDs, before/after values
and material destructive scope. Bind destructive work to a content revision or
immutable object-version scope, including its fingerprint in public evidence.
Apply must enforce that scope with conditional writes and refuse a nonempty
bucket delete. Leave out secrets, timestamps and unrelated volatile counts.
If the planned work changes, approval must be obtained again. Private `Plan.data`
never enters the public assessment; the checked frozen input model and the same
Plan reach apply.

Custom plans must be read-only. Apply cannot request approval: a late
`ApprovalRequired` becomes `late_gate_contract_violation` with uncertain state.
Keep live-state-dependent authorization wholly in the custom plan when shared
rules over model defaults would check different values from those executed.

## Progress, cancellation and results

Stages use `<system>.<verb>` keys. Declared `stages` determine 1-based position;
repeat and skip them freely. Progress counts are monotonic within a stage. The
first and final counts are sent, intermediate updates are limited to one every
two seconds, and the latest suppressed count is flushed on stage exit. Evidence
is a JSON-safe mapping under 2 KiB without reserved progress keys. Offline calls
log progress. Delivery failures do not fail backend work.

`cancellable=True` is cooperative. `cancel_task` accepts a request; only
`ctx.checkpoint()` stops work. Call it only after prior backend operations have
settled and the current state is safe to leave. Stage entry and progress never
stop work. A request after the last checkpoint can finish normally.

Use `Done` for ordinary success or return a full `ProviderResult` for advanced
outcomes (`pending_retry`, provenance, mixed resources). Advanced results retain
all their fields. Record each survivor with `ctx.created`, specifying `kind=` for
mixed kinds, and call `ctx.removed` only after confirmed cleanup. Survivors are
merged into success, typed errors, unexpected failures and cancellation results.
There is no rollback or durable SDK journal. A retried task must rediscover its
resources through provider-owned backend identity.

## Manifest subset

v1 supports scalar `str`, `int`, `float`, `bool`, scalar `Literal`, defaults,
integer `ge`/`le`, string `pattern`, title/description, canonical `Field(alias=...)`,
`ResourceRef`, and field extras `unit`, `group`, `semantic`. Action models must be
frozen. Unsupported constraints, models, serializers, validators and metadata
fail startup. Nested models, unions, lists and secret types are outside v1.

Actions expose description, target, destructive, fields and optional stages.
Resource declarations expose list fields and filters. Exact `get` is optional;
without one the SDK walks list cursors until the native reference is found.
Undeclared row fields are dropped while downloads, timestamps and other native
metadata are retained. `Page` is the contracts `ResourceListResponse`: pass its
required `observed_at` freshness timestamp. `App.list_resources` returns the
existing runtime mapping shape; list callbacks return `Page`.

There are no capability overlays, native backend cancellation, console/build-log
callbacks, structured inputs, placement engines or image workflows in v1.
Providers with those needs can retain their own Provider implementation.

## Development and snapshots

Install the released contracts/runtime dependencies and SDK development tools:

```bash
uv venv --python 3.13
uv pip install -e '.[dev]'
.venv/bin/ruff check .
.venv/bin/ty check src tests
.venv/bin/pytest
.venv/bin/python -m osp_provider_sdk manifest examples.buckets:app --check examples/manifest.json
.venv/bin/python -m build
.venv/bin/twine check dist/*
```

Generate snapshots with `python -m osp_provider_sdk manifest module:app`.
`--check [path]` compares exact canonical JSON, defaulting to `manifest.json`.
Snapshot generation checks startup but never enters lifespan or connects a backend.

CI runs the SDK and shared provider conformance suite with lowest-direct and
highest dependency resolution, checks the manifest, builds both artifacts,
and imports the installed wheel outside the source directory. Tag CI attaches
artifacts to a GitHub release; it does not publish to PyPI.

## First release

Release the orchestrator fingerprint lifecycle first, then contracts 0.9.x,
runtime 0.10.x, and SDK 0.1.0. Contracts must contain Plan/Gate, cleared
fingerprints, resource access and request-policy exports; runtime must expose
`unexpected_provider_error`. Re-run both dependency-resolution CI lanes with
those released versions. Finish the plan's field-vocabulary prerequisite,
NREC/tutorial pilot and Q1 audit before declaring adoption ready. Existing
providers and cookiecutter tests stay in place.

The private repository's Actions policy must allow its configured actions.
Set up the new distribution name and PyPI credentials or trusted publishing.
After review, quality checks and signed release/tag verification, a maintainer
can run `bash scripts/publish.sh` (or `-r testpypi`). `--tag` additionally creates,
verifies and pushes a signed tag after upload; normal development never runs it.
No release or live environment is needed for the local tests.
