Runtime invariants

The plugin-system guarantees that the same plugin behaves identically:

in an in-process application (FastAPI, Django, an ordinary Python/Node/Go service);
inside an orchestrator op (Dagster asset, Airflow task);
in a task-queue worker (Celery, RQ);
in a unit test (with mocked resources).

That property is not accidental. It only holds when the plugin observes eight runtime invariants fixed in ADR-0003. If even one is violated, the plugin stops working when it crosses from one environment to another.

This page is an overview of how each invariant is applied in practice. The normative contract lives in ADR-0003.

Summary table

#	Invariant	How it is checked
1	Orchestration neutrality (no reliance on ambient host state)	Automated test across three hosts.
2	Serialisable boundaries	Round-trip JSON serialisation for every input and output.
3	Resources via DI	Verifying that the plugin does not construct HTTP/DB clients itself.
4	Explicit `execution_model`	Manifest declaration; runtime detector for blocking I/O in sync plugins.
5	Unit of Work for long-running plugins	Manifest declaration.
6	Abstract ProgressSink / CheckpointStore	Absence of direct WebSocket / file I/O imports.
7	Idempotency modes (`input_hash` / `output_hash` / `none`)	Manifest declaration plus a re-run with the same input.
8	Determinism for `output_hash`	Two runs with a frozen clock and seeded RNG, bit-equal comparison.

1. Orchestration neutrality

Rule: a plugin must not rely on ambient host state — the current event loop, process-level singletons, locale, or working directory.

What is forbidden:

Relying on asyncio.get_event_loop() / Tokio::current() / equivalents in the constructor or setup().
Reading or writing process-level singletons (thread-locals, module-mutable state, a "global client").
Using the file system outside ctx.resources.tmpdir.
Reading os.environ during the runtime phase (reading it in setup() and caching the result is fine).
Relying on os.getcwd().

How to fix violations:

Python
TypeScript
Go

# ✗ Bad — capturing an external event loop.
class BadPlugin:
    def __init__(self):
        self.loop = asyncio.get_event_loop()  # breaks inside an orchestrator

# ✓ Good — never hold a reference to the loop.
class GoodPlugin:
    def __init__(self):
        pass  # lazy-init; everything happens in setup()

    async def setup(self, ctx):
        # asyncio.get_running_loop() inside an async function returns the current run's loop — fine.
        pass

:::warning TypeScript runtime ships in Phase 1 @dagstack/plugin-system@0.1.0-rc.2 exports only the spec-emitted types — VERSION, ToolV1, OrchestratorV1. The runtime (PluginRegistry, discover, dispatchers, contract suite) lands in Phase 1. Today: implement the kind contract against the published types, then host plugins through Python over mcp_stdio or wait for the Phase 1 release. See the TypeScript API reference for the planned shape. :::

// ✗ Bad — global state.
var cachedPool *sql.DB

// ✓ Good — through Resources DI.
type GoodPlugin struct {
    db DBClient // user-defined interface
}

func (p *GoodPlugin) Unwrap() any { return p }

func (p *GoodPlugin) Setup(ctx context.Context, pluginCtx *pluginsystem.PluginContext) error {
    raw, err := pluginCtx.Resources.Get("postgres")
    if err != nil {
        return err
    }
    p.db = raw.(DBClient)
    return nil
}

2. Serialisable boundaries

Rule: anything that crosses the plugin boundary (hook input, hook output, checkpoint, progress event) must be JSON-serialisable.

What may not be passed:

Live HTTP/DB clients, session objects, connection pools.
Open file descriptors.
Async generators, coroutines, threads, channels.
Closures and bound methods.
C-extension native handles.

What is allowed:

Structured data describable by JSON Schema.
Primitives (strings, numbers, booleans, null).
Arrays and dictionaries thereof.
Bytes (up to ≤10 MB in the v1.0 MVP).
References to external resources expressed as URLs: s3://bucket/key, file://path.

Common "silently breaks JSON" cases:

Type	How to serialise
`datetime`	ISO 8601 string (`"2026-01-01T12:00:00Z"`)
`Decimal` / `BigInt`	string
`set`	array
`Enum`	enum value
`UUID`	string (`"a1b2c3..."`)

Exception for in_process_only plugins: live objects (for example, stream objects) may be passed within a single hook invocation. Checkpoints and progress events, however, must remain serialisable.

3. Resources via DI

Rule: HTTP clients, database connections, and connection pools are injected through ctx.resources; they are not constructed inside the plugin.

The details live on a dedicated page, Resources (Resources DI).

4. Explicit `execution_model`

Rule: the plugin declares its execution style in the manifest. The host picks an appropriate executor.

execution_model = "async"
# or
execution_model = "sync"
execution_model = "thread_cpu_bound"
execution_model = "process_cpu_bound"

Consequences:

A sync plugin must not block on I/O — the host verifies this with specialised tooling (in Python, blockbuster).
An async plugin must not occupy the CPU without an explicit yield — that blocks the event loop.
A thread_cpu_bound plugin runs on a thread pool, not on the event loop.
process_cpu_bound — a separate process (for very heavy tasks).

5. Unit of Work for long-running plugins

Rule: a plugin whose operation takes minutes or more declares its UoW parameters in the manifest:

[plugin.unit_of_work]
declared = true
partition_key = "tenant_id"
estimated_duration_sec = 600
idempotency_mode = "input_hash"
checkpointable = true

partition_key — sharding key; the orchestrator parallelises units along this key.
idempotency_mode — see invariant 7.
checkpointable = true — the plugin persists progress through ctx.checkpoint and supports resume.

UoW plugins are invoked by the orchestrator (kind = "orchestrator"), not directly by the application.

6. Abstract ProgressSink / CheckpointStore

Rule: a plugin publishes progress through the abstract ctx.progress, not directly through WebSocket / SSE / a log file. The same applies to ctx.checkpoint for persisting and restoring state.

Python
TypeScript
Go

# ✗ Bad — a direct dependency on WebSocket.
class BadPlugin:
    async def process(self):
        await broadcast_to_websocket({"event": "progress", "percent": 50})

# ✓ Good — through the abstract sink.
class GoodPlugin:
    async def process(self):
        self._progress.update(percent=0.5, message="Halfway there")

func (p *GoodPlugin) Process() error {
    p.progress.Update(0.5, "Halfway there", nil)
    return nil
}

The host plugs in different sink implementations:

In a web application — a WebSocket broadcast.
In an orchestrator op — a stream into the orchestrator's state.
In a unit test — appending to an array for assertions.

7. Idempotency modes

Rule: a UoW plugin declares its idempotency_mode:

Mode	Semantics	When to use it
`input_hash`	A repeat with the same input → the same output. The orchestrator skips the repeat.	Indexing, data enrichment.
`output_hash`	A repeat may produce a new output, but if the hash matches it counts as a single result.	Idempotent downstream updates by result hash.
`none`	Every run is a new result. Deduplication is the consumer's responsibility.	Notifications, fire-and-forget.

8. Determinism for `output_hash`

Rule: a plugin with idempotency_mode = "output_hash" must be deterministic — the same input plus the same resource state yields the same output.

What is forbidden:

time.now() / Date.now() / time.Now() — only ctx.resources.clock.now().
random() / Math.random() / rand.Int() — only ctx.resources.rng.next(...).
Iterating a set without sorting.
Iterating an unordered dict / map.
Locale-dependent formatting.

How to verify determinism: run the plugin twice with a frozen clock and a seeded RNG, then compare outputs byte by byte. If they differ, some source of non-determinism is bypassing DI.

How the contract test framework checks the invariants

The Python binding (dagstack-plugin-system >= 0.2) ships a built-in contract test framework. Equivalent APIs in the TypeScript and Go implementations will land once they reach stable. A Python example follows:

from dagstack.plugin_system import (
    run_contract_suite,
    assert_no_ambient_state,
    assert_json_serializable_boundaries,
    assert_lifecycle_clean,
    assert_deterministic,
    assert_manifest_valid,
    ALL_CHECKS,
)


def test_my_plugin_contracts():
    manifest = load_manifest("plugins/my-plugin/dagstack.toml")
    result = run_contract_suite(
        plugin_class=MyPlugin,
        manifest=manifest,
        checks=ALL_CHECKS,
    )
    assert result.ok, result.format_failures()

Each check covers one or more invariants:

Check	Invariants
`assert_manifest_valid`	4, 5, 7 (manifest declarations)
`assert_no_ambient_state`	1
`assert_json_serializable_boundaries`	2
`assert_lifecycle_clean`	3 (Resources DI — no resource leaks)
`assert_deterministic`	8

Execution details live on the Guide: Testing plugins page.

Summary table​

1. Orchestration neutrality​

2. Serialisable boundaries​

3. Resources via DI​

4. Explicit execution_model​

5. Unit of Work for long-running plugins​

6. Abstract ProgressSink / CheckpointStore​

7. Idempotency modes​

8. Determinism for output_hash​

How the contract test framework checks the invariants​

See also​