diff --git a/README.md b/README.md
index d37ec15..c6a9abb 100644
--- a/README.md
+++ b/README.md
@@ -1,401 +1,140 @@
 <div align="center">
-    <h1><code>workers</code></h1>
-    <p>Bounded, composable worker primitives for the <a href="https://github.com/wago-org/wago">Wago</a> WebAssembly runtime.</p>
+  <h1><code>workers</code></h1>
+  <p>Bounded, composable WebAssembly worker primitives for Wago.</p>
 </div>
 
 <p align="center">
-    <a href="https://github.com/wago-org/workers/actions/workflows/ci.yml"><img src="https://github.com/wago-org/workers/actions/workflows/ci.yml/badge.svg" alt="CI"></a>
-    <a href="https://codecov.io/gh/wago-org/workers"><img src="https://codecov.io/gh/wago-org/workers/branch/main/graph/badge.svg" alt="Coverage"></a>
-    <a href="https://go.dev/"><img src="https://img.shields.io/badge/go-%3E%3D1.24-00ADD8.svg" alt="Go >= 1.24"></a>
-    <a href="https://github.com/wago-org/wago"><img src="https://img.shields.io/badge/wago-%3E%3D0.1.0-6E56CF.svg" alt="Wago >= 0.1.0"></a>
+  <a href="https://github.com/wago-org/workers/actions/workflows/ci.yml"><img src="https://github.com/wago-org/workers/actions/workflows/ci.yml/badge.svg" alt="CI"></a>
+  <a href="https://go.dev/"><img src="https://img.shields.io/badge/go-%3E%3D1.24-00ADD8.svg" alt="Go >= 1.24"></a>
 </p>
 
-<details>
-<summary>Table of Contents</summary>
+Workers turns a WebAssembly table entry into a bounded, host-supervised worker.
+Each worker is a managed fork of its caller, runs on one goroutine, and owns a
+fixed-capacity, byte-bounded mailbox.
 
-- [Overview](#overview)
-- [Installation](#installation)
-- [Concepts](#concepts)
-- [Usage](#usage)
-- [API](#api)
-  - [Spawning a worker](#spawning-a-worker)
-  - [Sending messages](#sending-messages)
-  - [Receiving inside a worker](#receiving-inside-a-worker)
-  - [Linking and termination](#linking-and-termination)
-  - [Worker options and limits](#worker-options-and-limits)
-  - [Errors](#errors)
-- [Examples](#examples)
-- [Testing](#testing)
-- [Architecture](#architecture)
-- [Contributing](#contributing)
-- [License](#license)
-- [Contact](#contact)
+The package deliberately stops at primitives: spawn, send, receive, link, kill,
+and observe. Policies such as supervision trees, restarts, and guest-visible
+mailbox ABIs belong in a plugin built on top.
 
-</details>
+> Workers is experimental (`v0.1.0`). Its API may change before the first stable
+> release.
 
-## Overview
-
-`workers` is an optional [Wago](https://github.com/wago-org/wago) plugin that turns a
-WebAssembly table entry into a **bounded, host-supervised worker**. Each worker is a
-forked instance of the caller, driven on its own goroutine, fed by a fixed-size,
-byte-bounded mailbox. The host observes every message and every exit.
-
-Workers does *not* define PIDs, guest-visible mailboxes, monitors, supervision trees, restart
-policy, or a guest ABI. Those are policies you build on top of the primitives here: `Spawn`, `Send`,
-`DispatchNext`, `Link`, and `Kill`, using the host callbacks `OnMessage` and `OnExit`.
-
-What you get out of the box:
-
-- **Bounded by construction**: every worker has a capped queue depth, a max payload
-  size, and a max total queued bytes. `Send` copies the payload and never blocks.
-- **Host-supervised**: `OnMessage` sees every delivered message; `OnExit` sees every
-  return, failure, or kill, with the cause.
-- **Lifecycle-safe**: a linked worker is torn down cooperatively when its creating
-  instance closes, and every worker is drained when the runtime shuts down.
-
-> **Stability:** experimental (`v0.0.0`). The API may change without notice.
-
-## Installation
-
-If you have the [`wago`](https://github.com/wago-org/wago) CLI installed:
-
-```sh
-wago pkg install github.com/wago-org/workers
-```
-
-or use [`go get`](https://pkg.go.dev/cmd/go#hdr-Get_packages_and_dependencies):
+## Install
 
 ```sh
-go get github.com/wago-org/workers
+wago add github.com/wago-org/workers
 ```
 
-The plugin requires two privileged capabilities - `instance.manage` (to fork and manage
-worker instances) and `instance.lifecycle` (to observe parent shutdown). Select the
-plugin in `wago.json`:
+The install review shows two exact required authorities:
 
-```json
-{
-  "$schema": "https://wago.sh/v0/schema.json",
-  "plugins": {
-    "wago-org/workers": "^0.0.0"
-  }
-}
-```
-
-Exact versions and reviewed authority belong in `wago-lock.json`:
-
-```json
-{
-  "plugins": {
-    "wago-org/workers": {
-      "version": "0.0.0",
-      "requiredCapabilities": [
-        "instance.manage",
-        "instance.lifecycle"
-      ],
-      "capabilities": {
-        "instance.manage": { "maxInstances": 64 },
-        "instance.lifecycle": true
-      }
-    }
-  }
-}
-```
-
-When you register the plugin programmatically, grant the same capabilities with
-[`wago.WithPluginGrants`](https://pkg.go.dev/github.com/wago-org/wago#WithPluginGrants)
-(see [Usage](#usage)). Grants are enforced in strict mode; omitting the option registers
-the plugin without capability checks.
+- `instance.manage`, with positive instance and memory ceilings, to fork and own
+  workers.
+- `instance.close.observe`, to stop linked workers when their exact creator
+  closes.
 
-## Concepts
+You may narrow the published `instance.manage` limits. The plugin also enforces
+its own worker-count and mailbox-memory ceilings.
 
-| Term | Meaning |
-| --- | --- |
-| **Plugin** | `workers.New()` - the `wago.Extension` you register with a runtime. It provides one **service**. |
-| **Service** (`*workers.Workers`) | The host-side handle you use to spawn, send, link, and kill workers, and to register `OnMessage` / `OnExit` observers. Obtain it with `plugin.Service()` or look it up through `workers.ServiceKey`. |
-| **Worker** | A forked child instance running one table entry with the exact Wasm signature `() -> ()`, on its own goroutine. Identified by a `WorkerID`. |
-| **Mailbox** | A fixed-capacity, byte-bounded FIFO queue owned by each worker. `Send` enqueues; the worker drains it cooperatively via `DispatchNext`. |
-| **Creator** | The instance that spawned the worker. `Link` ties a worker's lifetime to its exact creator. |
+## Use from another plugin
 
-A worker's life: **spawn → run its table entry → receive messages cooperatively →
-exit (return, fail, or kill) → `OnExit` fires → resources freed.**
-
-## Usage
-
-Register the plugin, grant its capabilities, and attach host observers:
+Workers provides the typed contract
+`github.com/wago-org/workers/service@1`. Declare it in the consumer's immutable
+definition:
 
 ```go
-package main
-
-import (
-	"log"
-
-	"github.com/wago-org/wago"
-	"github.com/wago-org/workers"
-)
-
-func main() {
-	workerPlugin := workers.New()
-
-	rt := wago.NewRuntime()
-	if err := rt.Use(workerPlugin, wago.WithPluginGrants(
-		wago.PluginManagedInstances, // "instance.manage"
-		wago.PluginInstanceHooks,    // "instance.lifecycle"
-	)); err != nil {
-		log.Fatal(err)
+var WorkersContract = workers.Contract
+
+func Definition() wago.PluginDefinition {
+	return wago.PluginDefinition{
+		// ...ID, version, and provenance...
+		Requires: []wago.PluginRequirement{{
+			ID: workers.PluginID, Version: "^0.1.0",
+		}},
+		Consumes: []wago.ContractRequirement{{
+			ID: WorkersContract.ID(), Major: WorkersContract.Major(),
+			Mode: wago.ContractRequired,
+		}},
 	}
-	defer rt.Close()
-
-	service := workerPlugin.Service()
-
-	// Observe every message delivered to any worker.
-	service.OnMessage(func(ctx *workers.MessageContext) error {
-		// Decode ctx.Tag / ctx.Payload, or write into ctx.Caller.Memory().
-		// Returning an error stops that worker with WorkerFailed.
-		return nil
-	})
-
-	// Observe every worker exit: return, failure, or kill.
-	service.OnExit(func(ctx *workers.WorkerExitContext) {
-		// Build monitoring, restarts, signalling, or logging here.
-		log.Printf("worker %d exited: kind=%d err=%v", ctx.WorkerID, ctx.Kind, ctx.Err)
-	})
-
-	// ... compile and instantiate guest modules that spawn/drive workers.
 }
 ```
 
-`Service()` returns the same handle that a downstream plugin can retrieve through the
-exported `workers.ServiceKey`, so composing plugins share one worker registry. Inside your
-plugin's `Register`, take a typed reference and resolve it once the runtime is wired up:
+Require the contract during registration and use it only through its callback:
 
 ```go
-ref, err := plugin.Require(reg, workers.ServiceKey)
+workersRef, err := wagoplugin.Require(reg, workers.Contract)
 if err != nil {
 	return err
 }
-// later, when handling a call:
-svc, err := ref.Get()
-```
-
-## API
-
-All host-side operations that cross into a guest - `Spawn`, `DispatchNext`, `Current`,
-`Link` - must be called from **inside a synchronous host import**, where `caller` is the
-active `wago.HostModule` for that call. `Send` and `Kill` take a `WorkerID` and may be
-called from anywhere.
-
-### Spawning a worker
 
-Spawn forks the calling instance and runs the table entry at `tableIndex`. The entry must
-have the exact Wasm signature `() -> ()`; otherwise `Spawn` returns a validation error and
-the child is discarded.
-
-```go
-id, err := service.Spawn(caller, tableIndex, workers.WorkerOptions{
-	QueueCapacity:   64,
-	MaxPayloadBytes: 64 << 10, // 64 KiB
-	MaxQueueBytes:   1 << 20,  // 1 MiB
+err = workersRef.With(func(service workers.Service) error {
+	id, err := service.Spawn(caller, tableIndex, workers.WorkerOptions{})
+	if err != nil {
+		return err
+	}
+	return service.Send(id, 42, payload)
 })
-if err != nil {
-	return err
-}
 ```
 
-Zero-valued options fall back to the package defaults (see
-[Worker options and limits](#worker-options-and-limits)).
+Wago records the exact provider binding in `wago-lock.json`. A new matching
+provider cannot silently enter the graph, and the reference fails closed after
+the consumer stops. The package requirement lets the resolver install Workers
+transitively; the contract requirement records the exact reviewed call target.
 
-### Sending messages
+## Observe safely
 
-`Send` copies the payload and returns immediately - it never blocks the caller. It fails
-if the worker is unknown, stopping, or its mailbox is at capacity (by count or by bytes),
-or if the payload exceeds the worker's `MaxPayloadBytes`.
+Observers return an opaque subscription token:
 
 ```go
-if err := service.Send(id, 42 /* tag */, []byte("hello")); err != nil {
-	// e.g. errors.Is(err, workers.ErrWorkerQueueFull)
-	return err
-}
-```
-
-Because the payload is copied, the caller may reuse or mutate its buffer as soon as `Send`
-returns.
-
-### Receiving inside a worker
-
-A worker drains one message per `DispatchNext`, which you expose to the guest through a
-plugin-defined host import. It delivers the next queued message to every registered
-`OnMessage` observer, blocking cooperatively until a message arrives or the worker stops.
-
-```go
-// Inside the host import the guest calls to receive:
-if err := service.DispatchNext(caller); err != nil {
-	return pluginErrno(err) // map to a guest-visible errno
-}
-```
-
-A worker can learn its own ID with `Current`:
-
-```go
-id, err := service.Current(caller)
-```
-
-### Linking and termination
-
-`Link` ties a worker's lifetime to its **exact creator**: when the creating instance
-closes, the runtime's `BeforeClose` hook cooperatively stops every linked worker and waits
-for them to drain. A worker cannot link to itself, and only the creator may link it.
-
-```go
-if err := service.Link(caller, childID); err != nil {
+err = workersRef.With(func(service workers.Service) error {
+	messages, err = service.ObserveMessages(func(ctx *workers.MessageContext) error {
+		return handleMessage(ctx.WorkerID, ctx.Tag, ctx.Payload)
+	})
 	return err
-}
-```
-
-`Kill` requests cooperative termination of any worker by ID. The worker's in-flight
-`DispatchNext` unblocks, its queue is cleared, and it exits with `WorkerKilled`.
-
-```go
-_ = service.Kill(id)
-```
-
-Every exit - whether the table entry returned, an `OnMessage` observer returned an error,
-or the worker was killed - surfaces through `OnExit`:
-
-```go
-service.OnExit(func(ctx *workers.WorkerExitContext) {
-	switch ctx.Kind {
-	case workers.WorkerReturned: // clean return from the table entry
-	case workers.WorkerFailed:   // trap or observer error; see ctx.Err
-	case workers.WorkerKilled:   // Kill, parent close, or runtime shutdown
-	}
 })
 ```
 
-### Worker options and limits
-
-`WorkerOptions` bounds a single worker. A zero field takes the package default. Values
-above the hard maximum (or a `MaxQueueBytes` smaller than `MaxPayloadBytes`) are rejected
-with `ErrInvalidWorkerOptions`.
+Pass subscriptions to `service.Unsubscribe` through `workersRef.With` in the
+consuming plugin's `Stop` callback. Unsubscribe removes the observer and waits
+for callbacks already in flight, so the consumer can then release its state
+safely. It must not be called from inside its own callback.
 
-| Field | Default | Maximum | Meaning |
-| --- | --- | --- | --- |
-| `QueueCapacity` | `64` | `65536` | Max number of queued messages. |
-| `MaxPayloadBytes` | `64 KiB` | `16 MiB` | Max size of a single `Send` payload. |
-| `MaxQueueBytes` | `1 MiB` | `64 MiB` | Max total bytes queued at once (must be ≥ `MaxPayloadBytes`). |
+## Worker operations
 
-The corresponding exported constants are `DefaultWorkerQueueCapacity`,
-`DefaultWorkerMaxPayloadBytes`, `DefaultWorkerMaxQueueBytes`, `MaxWorkerQueueCapacity`,
-`MaxWorkerPayloadBytes`, and `MaxWorkerQueueBytes`.
+`Spawn`, `Current`, `DispatchNext`, and `Link` must run inside a synchronous host
+call and receive that call's `wago.HostModule`. This prevents one guest from
+impersonating another.
 
-`WorkerOptions` bounds one worker; `WorkerLimits` bounds the **whole service**, so a
-guest cannot exhaust the host by spawning workers without end. It always applies — even
-when the host grants `instance.manage` with no `maxInstances` budget — and complements
-that core budget rather than replacing it. Pass it at construction:
+- `Spawn` forks the caller and runs a `() -> ()` table entry.
+- `Send` copies a tagged payload into a bounded mailbox and never blocks.
+- `DispatchNext` waits cooperatively and delivers one queued message. Its
+  context lets the consuming plugin cancel the wait during `Stop`.
+- `Current` returns the worker ID for the current managed caller.
+- `Link` ties a worker to its exact creator.
+- `Kill` requests cooperative termination by worker ID.
 
-```go
-workerPlugin := workers.New(workers.WithLimits(workers.WorkerLimits{
-	MaxLiveWorkers: 128,       // default 64
-	MaxQueueBytes:  128 << 20, // default 64 MiB, summed across live workers
-}))
-```
+Every worker has `QueueCapacity`, `MaxPayloadBytes`, and `MaxQueueBytes` bounds.
+Zero fields take safe package defaults.
 
-| Field | Default | Meaning |
-| --- | --- | --- |
-| `MaxLiveWorkers` | `64` | Max workers live at once. `Spawn` returns `ErrWorkerQuotaExceeded` past it. |
-| `MaxQueueBytes` | `64 MiB` | Max total per-worker `MaxQueueBytes` reservation summed across live workers. |
-
-### Errors
-
-All errors are comparable sentinels - match them with `errors.Is`.
-
-| Error | When |
-| --- | --- |
-| `ErrWorkersInactive` | Service is nil or not registered. |
-| `ErrInvalidWorkerOptions` | Options exceed a hard limit or are inconsistent. |
-| `ErrInvalidWorkerCaller` | Operation needs the current plugin host caller and none is active. |
-| `ErrWorkerImportLifetime` | The worker would inherit a borrowed import it cannot safely keep. |
-| `ErrWorkerNotFound` | No worker with that ID (or not owned by the caller). |
-| `ErrWorkerStopping` | The worker is already shutting down. |
-| `ErrWorkerQueueFull` | Mailbox at capacity by count or bytes. |
-| `ErrWorkerDispatchActive` | `DispatchNext` is already running for that worker. |
-| `ErrPayloadTooLarge` | Payload exceeds the worker's `MaxPayloadBytes`. |
-| `ErrWorkerIDExhausted` | The `WorkerID` space is exhausted. |
-| `ErrInvalidWorkerLink` | Link target is not the caller's direct child, or is self. |
-| `ErrWorkerKilled` | Exit cause: `Kill` was requested. |
-| `ErrWorkerParentClosed` | Exit cause: the linked creator instance closed. |
-| `ErrWorkerRuntimeClosed` | Exit cause: the runtime shut down. |
-| `ErrWorkerQuotaExceeded` | Spawn would exceed the service-wide `WorkerLimits`. |
-
-## Examples
-
-The end-to-end integration test is the canonical, runnable example: it defines a guest
-module that imports `spawn`/`next`, spawns a worker, and verifies the host copies a
-message and observes a clean exit. See
-[`workers_test.go`](./workers_test.go) - in particular `TestPluginSpawnsCopiesMessageAndStops`.
-
-A generated Wago host does not import this package directly; it blank-imports the
-[`register`](./register) subpackage, which activates the plugin's init-time registration:
-
-```go
-import _ "github.com/wago-org/workers/register"
-```
-
-## Testing
+## Configure
 
 ```sh
-go test ./...
+wago plugin config github.com/wago-org/workers \
+  '{"maxLiveWorkers":128,"maxQueueBytes":134217728}'
 ```
 
-The suite uses `github.com/wago-org/wago/testutil/wasmtest` to hand-build guest modules,
-so no external toolchain is required. Run with the race detector while iterating on the
-concurrency paths:
+Configuration is strict: unknown fields, zero explicit limits, out-of-range
+values, and trailing JSON are rejected before activation.
+
+## Test
 
 ```sh
+go test ./...
 go test -race ./...
 ```
 
-## Architecture
-
-- **`workers.go`** - the whole plugin: the `Plugin` extension, the `Workers` service, the
-  per-worker mailbox and goroutine, and the lifecycle wiring (`BeforeClose` teardown,
-  runtime-close drain).
-- **`register/`** - a blank-import shim that activates init-time registration for
-  Wago-generated hosts.
-- **`wago.json`** - the package manifest declaring plugin dependencies and
-  version constraints; reviewed capabilities live in `wago-lock.json`.
-
-Design notes:
-
-- Each worker runs on one goroutine (`worker.run`) that invokes the table entry and, on
-  return, resolves the exit kind, closes the child instance, unregisters it, and fans out
-  to `OnExit` observers (each guarded against panics).
-- The mailbox is a fixed-size ring buffer guarded by a mutex; `Send` enqueues and signals,
-  `DispatchNext` dequeues and delivers. Back-pressure is explicit: a full queue rejects
-  rather than blocks.
-- `Spawn` forks through the managed-instance API and validates the target table entry
-  before the worker is made visible, so a bad spawn leaves no partial state.
-
-## Contributing
-
-Contributions are welcome! Please:
-
-- Run `go test -race ./...` and `go vet ./...` before opening a pull request.
-- Keep the plugin's boundaries intact - no PIDs, guest mailboxes, supervision, or guest
-  ABI belong in this package; those are policies for layers built on top.
-- Follow standard Go formatting (`gofmt`) and conventional commit messages.
+The suite covers quotas, host-call identity, worker teardown, copied payloads,
+typed contract revocation, and observer unsubscribe races.
 
 ## License
 
-This project is distributed under the [Apache License 2.0](./LICENSE). Work on this
-project is done out of passion - if you want to support it financially, you can donate
-through [GitHub Sponsors](https://github.com/sponsors/JairusSW).
-
-## Contact
-
-Please file issues at [GitHub Issues](https://github.com/wago-org/workers/issues). To chat,
-join the [Wago Discord](https://wago.sh/discord).
-
-- **GitHub:** [https://github.com/wago-org/](https://github.com/wago-org/)
-- **Website:** [https://wago.sh/](https://wago.sh/)
-- **Discord:** [https://wago.sh/discord](https://wago.sh/discord)
+Apache-2.0. See [LICENSE](./LICENSE).
diff --git a/go.mod b/go.mod
index 247fc55..82d9338 100644
--- a/go.mod
+++ b/go.mod
@@ -2,4 +2,4 @@ module github.com/wago-org/workers
 
 go 1.24.0
 
-require github.com/wago-org/wago v0.0.0-20260711053856-7d89b98c854d
+require github.com/wago-org/wago v0.0.0-20260812144524-1c58c9862d25
diff --git a/go.sum b/go.sum
index ff18b21..24b3655 100644
--- a/go.sum
+++ b/go.sum
@@ -1,2 +1,2 @@
-github.com/wago-org/wago v0.0.0-20260711053856-7d89b98c854d h1:fzJn6UlsGwiqwyTN+Si6oFoETUhUoJ/Rphhx94M5Pqc=
-github.com/wago-org/wago v0.0.0-20260711053856-7d89b98c854d/go.mod h1:d44S+59u6VHeqEaIZBKTeB8viyswTP97O9IKA6Gyzc4=
+github.com/wago-org/wago v0.0.0-20260812144524-1c58c9862d25 h1:gopNHvaUaDSzmW2Hfd32r8dOwwzYUHH3ph1L/IuhJUA=
+github.com/wago-org/wago v0.0.0-20260812144524-1c58c9862d25/go.mod h1:6XmxI3S5qJ+YAzObyXjC6FmxHtbLhBBxrglzix42zl8=
diff --git a/manifest_test.go b/manifest_test.go
new file mode 100644
index 0000000..2a1720f
--- /dev/null
+++ b/manifest_test.go
@@ -0,0 +1,100 @@
+package workers_test
+
+import (
+	"encoding/json"
+	"os"
+	"reflect"
+	"testing"
+
+	"github.com/wago-org/wago"
+	"github.com/wago-org/workers"
+	workersregister "github.com/wago-org/workers/register"
+)
+
+type manifestAuthor struct {
+	Name string `json:"name"`
+}
+
+type manifestPackage struct {
+	Module      string            `json:"module"`
+	Version     string            `json:"version"`
+	Name        string            `json:"name"`
+	Description string            `json:"description"`
+	Stability   wago.Stability    `json:"stability"`
+	License     string            `json:"license"`
+	Homepage    string            `json:"homepage"`
+	Repository  string            `json:"repository"`
+	Authors     []manifestAuthor  `json:"authors"`
+	Engines     map[string]string `json:"engines"`
+	Platforms   []string          `json:"platforms"`
+}
+
+func TestManifestMatchesCatalogMetadata(t *testing.T) {
+	raw, err := os.ReadFile("wago.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var manifest struct {
+		Schema  string            `json:"$schema"`
+		Package manifestPackage   `json:"package"`
+		Plugins map[string]string `json:"plugins"`
+	}
+	if err := json.Unmarshal(raw, &manifest); err != nil {
+		t.Fatal(err)
+	}
+	if manifest.Schema != "https://wago.sh/v1/schema.json" {
+		t.Fatalf("manifest schema = %q", manifest.Schema)
+	}
+	providers := workersregister.Providers()
+	if len(providers) != 1 {
+		t.Fatalf("catalog providers = %d, want 1", len(providers))
+	}
+	definition := workers.Definition()
+	if !reflect.DeepEqual(providers[0].Definition, definition) {
+		t.Fatalf("catalog definition drifted\ncatalog=%#v\ncanonical=%#v", providers[0].Definition, definition)
+	}
+	assertManifestMetadata(t, manifest.Package, definition)
+	if len(manifest.Plugins) != 0 {
+		t.Fatalf("leaf manifest dependencies = %v", manifest.Plugins)
+	}
+	assertProviderCatalogCurrent(t, "github.com/wago-org/workers/register", providers)
+}
+
+func assertProviderCatalogCurrent(t *testing.T, importPath string, providers []wago.PluginProvider) {
+	t.Helper()
+	want, err := wago.EncodeProviderCatalog(importPath, providers)
+	if err != nil {
+		t.Fatal(err)
+	}
+	got, err := os.ReadFile(wago.ProviderCatalogFile)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("%s is stale; run wago plugin catalog", wago.ProviderCatalogFile)
+	}
+	if _, err := wago.DecodeProviderCatalog(got); err != nil {
+		t.Fatalf("%s: %v", wago.ProviderCatalogFile, err)
+	}
+}
+
+func assertManifestMetadata(t *testing.T, manifest manifestPackage, definition wago.PluginDefinition) {
+	t.Helper()
+	authors := make([]string, len(manifest.Authors))
+	for i := range manifest.Authors {
+		authors[i] = manifest.Authors[i].Name
+	}
+	if manifest.Module != definition.ID ||
+		manifest.Version != definition.Version ||
+		manifest.Name != definition.Name ||
+		manifest.Description != definition.Description ||
+		manifest.Stability != definition.Stability ||
+		manifest.License != definition.Provenance.License ||
+		manifest.Homepage != definition.Provenance.Homepage ||
+		manifest.Repository != definition.Provenance.Repository ||
+		!reflect.DeepEqual(authors, definition.Provenance.Authors) ||
+		!reflect.DeepEqual(manifest.Engines, definition.Compatibility.Engines) ||
+		!reflect.DeepEqual(manifest.Platforms, definition.Compatibility.Platforms) {
+		t.Fatalf("manifest metadata drifted\nmanifest=%#v\ndefinition=%#v", manifest, definition)
+	}
+}
diff --git a/register/catalog.go b/register/catalog.go
new file mode 100644
index 0000000..d314dd5
--- /dev/null
+++ b/register/catalog.go
@@ -0,0 +1,12 @@
+// Package register exposes Workers' explicit provider catalog to generated Wago
+// runtimes. Importing this package has no registration side effects.
+package register
+
+import (
+	"github.com/wago-org/wago"
+	"github.com/wago-org/workers"
+)
+
+func Providers() []wago.PluginProvider {
+	return []wago.PluginProvider{workers.Provider()}
+}
diff --git a/register/register.go b/register/register.go
deleted file mode 100644
index 62ef0c9..0000000
--- a/register/register.go
+++ /dev/null
@@ -1,6 +0,0 @@
-// Package register activates the workers plugin's init-time registration for
-// Wago-generated hosts. Applications normally import github.com/wago-org/workers
-// directly; generated hosts blank-import this package.
-package register
-
-import _ "github.com/wago-org/workers"
diff --git a/wago.json b/wago.json
index d689173..a596a72 100644
--- a/wago.json
+++ b/wago.json
@@ -1,13 +1,17 @@
 {
-  "$schema": "https://wago.sh/v0/schema.json",
-  "module": "github.com/wago-org/workers",
-  "version": "0.0.0",
-  "name": "Wago Workers",
-  "short": "workers",
-  "description": "Bounded, composable WebAssembly worker primitives for Wago plugins.",
-  "license": "Apache-2.0",
-  "repository": "https://github.com/wago-org/workers",
-  "homepage": "https://github.com/wago-org/workers#readme",
-  "category": "concurrency",
-  "tags": ["workers", "concurrency", "webassembly"]
+  "$schema": "https://wago.sh/v1/schema.json",
+  "package": {
+    "module": "github.com/wago-org/workers",
+    "version": "0.1.0",
+    "name": "Workers",
+    "description": "Bounded, composable WebAssembly worker primitives for Wago plugins.",
+    "stability": "experimental",
+    "license": "Apache-2.0",
+    "homepage": "https://github.com/wago-org/workers#readme",
+    "repository": "https://github.com/wago-org/workers",
+    "category": "concurrency",
+    "tags": ["workers", "concurrency", "webassembly", "wasm"],
+    "authors": [{"name": "Wago contributors"}],
+    "engines": {"wago": ">=0.1.0"}
+  }
 }
diff --git a/wago.providers.json b/wago.providers.json
new file mode 100644
index 0000000..fb87993
--- /dev/null
+++ b/wago.providers.json
@@ -0,0 +1,68 @@
+{
+  "$schema": "https://wago.sh/v1/providers.schema.json",
+  "providers": [
+    {
+      "importPath": "github.com/wago-org/workers/register",
+      "definition": {
+        "id": "github.com/wago-org/workers",
+        "name": "Workers",
+        "version": "0.1.0",
+        "description": "Bounded, composable WebAssembly worker primitives for Wago plugins.",
+        "stability": "experimental",
+        "compatibility": {
+          "engines": {
+            "wago": "\u003e=0.1.0"
+          }
+        },
+        "provenance": {
+          "homepage": "https://github.com/wago-org/workers#readme",
+          "repository": "https://github.com/wago-org/workers",
+          "license": "Apache-2.0",
+          "authors": [
+            "Wago contributors"
+          ]
+        },
+        "authorities": [
+          {
+            "name": "instance.close.observe",
+            "mode": "required",
+            "reason": "stop linked workers when their exact creator closes",
+            "scope": {}
+          },
+          {
+            "name": "instance.manage",
+            "mode": "required",
+            "reason": "fork and own bounded worker instances",
+            "scope": {
+              "maxInstances": 1024,
+              "maxMemoryBytes": 4294967296
+            }
+          }
+        ],
+        "configSchema": {
+          "additionalProperties": false,
+          "properties": {
+            "maxLiveWorkers": {
+              "maximum": 65536,
+              "minimum": 1,
+              "type": "integer"
+            },
+            "maxQueueBytes": {
+              "maximum": 68719476736,
+              "minimum": 1,
+              "type": "integer"
+            }
+          },
+          "type": "object"
+        },
+        "provides": [
+          {
+            "id": "github.com/wago-org/workers/service",
+            "major": 1
+          }
+        ]
+      },
+      "definitionDigest": "sha256:efe2db34dbba3b2782a5ec3fbc741945bb98c2d5ddb20d127b0fa31de4fb3425"
+    }
+  ]
+}
diff --git a/workers.go b/workers.go
index 29ef47f..84f6874 100644
--- a/workers.go
+++ b/workers.go
@@ -2,16 +2,20 @@
 package workers
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"sort"
 	"sync"
 
 	"github.com/wago-org/wago"
-	"github.com/wago-org/wago/plugin"
+	wagoplugin "github.com/wago-org/wago/plugin"
 )
 
-const PluginName = "workers"
+const PluginID = "github.com/wago-org/workers"
 
 type WorkerID uint64
 
@@ -26,8 +30,8 @@ const (
 	// DefaultMaxLiveWorkers bounds the number of simultaneously live workers for
 	// one service when WorkerLimits.MaxLiveWorkers is left zero. Each worker owns a
 	// managed instance, a goroutine, and a foreign stack, so an unbounded count is a
-	// denial-of-service vector; a default keeps Spawn bounded even when the host
-	// grants instance.manage without a maxInstances budget.
+	// denial-of-service vector; a package-level default remains useful even when
+	// the reviewed instance.manage grant permits a larger ceiling.
 	DefaultMaxLiveWorkers uint32 = 64
 	// DefaultMaxServiceQueueBytes bounds the total queued-payload reservation across
 	// all live workers when WorkerLimits.MaxQueueBytes is left zero.
@@ -107,71 +111,189 @@ type WorkerExitContext struct {
 	Err      error
 }
 
-var ServiceKey = plugin.NewServiceKey[*Workers]("wago.workers/v1")
-
-type Plugin struct {
-	service *Workers
-	limits  WorkerLimits
+// Service is Workers' typed cross-plugin contract. Call it only inside the
+// callback of a wagoplugin.Ref; Wago holds the provider alive for that callback.
+type Service interface {
+	Spawn(wago.HostModule, uint32, WorkerOptions) (WorkerID, error)
+	Send(WorkerID, uint64, []byte) error
+	Current(wago.HostModule) (WorkerID, error)
+	DispatchNext(context.Context, wago.HostModule) error
+	Link(wago.HostModule, WorkerID) error
+	Kill(WorkerID) error
+	ObserveMessages(func(*MessageContext) error) (Subscription, error)
+	ObserveExits(func(*WorkerExitContext)) (Subscription, error)
+	Unsubscribe(Subscription) error
+}
+
+// Contract is the major-versioned Workers composition seam.
+var Contract = wagoplugin.NewContract[Service](PluginID+"/service", 1)
+
+type pluginConfig struct {
+	MaxLiveWorkers *uint32 `json:"maxLiveWorkers,omitempty"`
+	MaxQueueBytes  *uint64 `json:"maxQueueBytes,omitempty"`
+}
+
+var configSchema = json.RawMessage(`{
+  "type": "object",
+  "additionalProperties": false,
+  "properties": {
+    "maxLiveWorkers": {"type": "integer", "minimum": 1, "maximum": 65536},
+    "maxQueueBytes": {"type": "integer", "minimum": 1, "maximum": 68719476736}
+  }
+}`)
+
+func decodePluginConfig(raw json.RawMessage) (pluginConfig, WorkerLimits, error) {
+	if len(raw) == 0 {
+		raw = json.RawMessage(`{}`)
+	}
+	if err := validateConfigObject(raw); err != nil {
+		return pluginConfig{}, WorkerLimits{}, fmt.Errorf("workers: config: %w", err)
+	}
+	var cfg pluginConfig
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&cfg); err != nil {
+		return pluginConfig{}, WorkerLimits{}, fmt.Errorf("workers: config: %w", err)
+	}
+	if err := dec.Decode(new(any)); err != io.EOF {
+		return pluginConfig{}, WorkerLimits{}, fmt.Errorf("workers: config has a trailing JSON value")
+	}
+	limits := WorkerLimits{MaxLiveWorkers: DefaultMaxLiveWorkers, MaxQueueBytes: DefaultMaxServiceQueueBytes}
+	if cfg.MaxLiveWorkers != nil {
+		if *cfg.MaxLiveWorkers == 0 || *cfg.MaxLiveWorkers > 65536 {
+			return pluginConfig{}, WorkerLimits{}, fmt.Errorf("workers: maxLiveWorkers must be in [1, 65536]")
+		}
+		limits.MaxLiveWorkers = *cfg.MaxLiveWorkers
+	}
+	if cfg.MaxQueueBytes != nil {
+		if *cfg.MaxQueueBytes == 0 || *cfg.MaxQueueBytes > 64<<30 {
+			return pluginConfig{}, WorkerLimits{}, fmt.Errorf("workers: maxQueueBytes must be in [1, 68719476736]")
+		}
+		limits.MaxQueueBytes = *cfg.MaxQueueBytes
+	}
+	return cfg, limits, nil
 }
 
-// Option configures the workers Plugin at construction.
-type Option func(*Plugin)
-
-// WithLimits sets the aggregate resource limits for the worker service. Zero
-// fields fall back to the package defaults (see WorkerLimits).
-func WithLimits(l WorkerLimits) Option { return func(p *Plugin) { p.limits = l } }
-
-// New creates the workers plugin. Pass WithLimits to override the default
-// aggregate resource caps.
-func New(opts ...Option) *Plugin {
-	p := &Plugin{}
-	for _, opt := range opts {
-		opt(p)
+func validateConfigObject(raw json.RawMessage) error {
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	token, err := dec.Token()
+	if err != nil {
+		return err
+	}
+	if token != json.Delim('{') {
+		return fmt.Errorf("must be a JSON object")
 	}
-	return p
+	seen := map[string]struct{}{}
+	for dec.More() {
+		keyToken, err := dec.Token()
+		if err != nil {
+			return err
+		}
+		key, ok := keyToken.(string)
+		if !ok {
+			return fmt.Errorf("object key is not a string")
+		}
+		if _, duplicate := seen[key]; duplicate {
+			return fmt.Errorf("duplicate field %q", key)
+		}
+		seen[key] = struct{}{}
+		var value json.RawMessage
+		if err := dec.Decode(&value); err != nil {
+			return err
+		}
+		if bytes.Equal(bytes.TrimSpace(value), []byte("null")) {
+			return fmt.Errorf("field %q must not be null", key)
+		}
+	}
+	if _, err := dec.Token(); err != nil {
+		return err
+	}
+	if err := dec.Decode(new(any)); err != io.EOF {
+		if err == nil {
+			return fmt.Errorf("has a trailing JSON value")
+		}
+		return err
+	}
+	return nil
 }
 
-func (*Plugin) Info() wago.ExtensionInfo {
-	return wago.ExtensionInfo{
-		ID: "wago.workers", Name: "Workers", Version: "0.0.0",
-		Description: "Bounded, extension-scoped WebAssembly worker primitives",
-		Stability:   wago.Experimental, Repository: "https://github.com/wago-org/workers",
-		License: "Apache-2.0", Tags: []string{"workers", "concurrency", "plugin-foundation"},
-		RequiresCapabilities: []wago.PluginCapability{wago.PluginManagedInstances, wago.PluginInstanceHooks},
-		Compat:               wago.Compatibility{Engines: map[string]string{"wago": ">=0.1.0"}},
+// Definition returns fresh immutable metadata for Workers' explicit provider.
+func Definition() wago.PluginDefinition {
+	return wago.PluginDefinition{
+		ID:          PluginID,
+		Name:        "Workers",
+		Version:     "0.1.0",
+		Description: "Bounded, composable WebAssembly worker primitives for Wago plugins.",
+		Stability:   wago.Experimental,
+		Compatibility: wago.Compatibility{
+			Engines: map[string]string{"wago": ">=0.1.0"},
+		},
+		Provenance: wago.PluginProvenance{
+			Homepage:   "https://github.com/wago-org/workers#readme",
+			Repository: "https://github.com/wago-org/workers",
+			License:    "Apache-2.0",
+			Authors:    []string{"Wago contributors"},
+		},
+		Authorities: []wago.AuthorityRequest{
+			{
+				Name: wago.AuthorityInstanceManage, Mode: wago.AuthorityRequired,
+				Reason: "fork and own bounded worker instances",
+				Scope:  wago.AuthorityScope{MaxInstances: 1024, MaxMemoryBytes: 4 << 30},
+			},
+			{
+				Name: wago.AuthorityInstanceCloseObserve, Mode: wago.AuthorityRequired,
+				Reason: "stop linked workers when their exact creator closes",
+			},
+		},
+		ConfigSchema: append(json.RawMessage(nil), configSchema...),
+		Provides:     []wago.ContractSpec{Contract.Spec()},
+	}
+}
+
+// Provider is Workers' side-effect-free catalog entry.
+func Provider() wago.PluginProvider {
+	return wago.PluginProvider{
+		Definition: Definition(),
+		New:        func() wago.Plugin { return new(plugin) },
+		ValidateConfig: func(raw json.RawMessage) error {
+			_, _, err := decodePluginConfig(raw)
+			return err
+		},
 	}
 }
 
-func (p *Plugin) Register(reg *wago.Registry) error {
+type plugin struct{ service *Workers }
+
+func (p *plugin) Register(reg *wago.Registrar) error {
+	var cfg pluginConfig
+	if err := reg.Config(&cfg); err != nil {
+		return err
+	}
+	limits := WorkerLimits{MaxLiveWorkers: DefaultMaxLiveWorkers, MaxQueueBytes: DefaultMaxServiceQueueBytes}
+	if cfg.MaxLiveWorkers != nil {
+		limits.MaxLiveWorkers = *cfg.MaxLiveWorkers
+	}
+	if cfg.MaxQueueBytes != nil {
+		limits.MaxQueueBytes = *cfg.MaxQueueBytes
+	}
 	manager, err := reg.ManagedInstances()
 	if err != nil {
 		return err
 	}
-	lifecycle, err := reg.InstanceLifecycle()
+	closeObserver, err := reg.InstanceCloseObserver()
 	if err != nil {
 		return err
 	}
-	p.service = newWorkers(manager, p.limits)
-	lifecycle.BeforeClose(func(ctx *wago.InstanceContext) { p.service.parentClosing(ctx.Instance) })
-	return plugin.Provide(reg, ServiceKey, p.service)
-}
-
-func (p *Plugin) Stop(context.Context) error {
-	if p == nil || p.service == nil {
-		return nil
+	p.service = newWorkers(manager, limits)
+	if err := closeObserver.Before(func(event wago.InstanceCloseEvent) { p.service.parentClosing(event.Instance) }); err != nil {
+		return err
 	}
-	return p.service.close()
-}
-
-func (p *Plugin) Service() *Workers {
-	if p == nil {
-		return nil
+	if err := wagoplugin.Provide(reg, Contract, Service(p.service)); err != nil {
+		return err
 	}
-	return p.service
+	return reg.Lifecycle(wago.PluginLifecycle{Stop: func(context.Context) error { return p.service.close() }})
 }
 
-func init() { wago.RegisterExtension(PluginName, func() wago.Extension { return New() }) }
-
 type Workers struct {
 	mu         sync.Mutex
 	manager    *wago.InstanceManager
@@ -180,16 +302,18 @@ type Workers struct {
 	queueBytes uint64 // total per-worker queue-byte reservation currently held
 	next       WorkerID
 	workers    map[WorkerID]*worker
-	byInstance map[*wago.Instance]*worker
-	messages   []func(*MessageContext) error
-	exits      []func(*WorkerExitContext)
+	byInstance map[wago.InstanceIdentity]*worker
+	nextObs    uint64
+	messages   map[uint64]*messageObserver
+	exits      map[uint64]*exitObserver
 	closed     bool
 	exitPanics []error
 }
 
 func newWorkers(manager *wago.InstanceManager, limits WorkerLimits) *Workers {
 	return &Workers{manager: manager, limits: normalizeLimits(limits), next: 1,
-		workers: map[WorkerID]*worker{}, byInstance: map[*wago.Instance]*worker{}}
+		workers: map[WorkerID]*worker{}, byInstance: map[wago.InstanceIdentity]*worker{},
+		messages: map[uint64]*messageObserver{}, exits: map[uint64]*exitObserver{}}
 }
 
 // reserve claims one live-worker slot and queueBytes of the aggregate queue-byte
@@ -217,19 +341,208 @@ func (w *Workers) release(queueBytes uint32) {
 	w.mu.Unlock()
 }
 
-func (w *Workers) OnMessage(fns ...func(*MessageContext) error) {
+type subscriptionKind uint8
+
+const (
+	messageSubscription subscriptionKind = iota + 1
+	exitSubscription
+)
+
+// Subscription is an opaque observer token. It has no provider operations of
+// its own; pass it back to Service.Unsubscribe inside a leased contract call.
+type Subscription struct {
+	id   uint64
+	kind subscriptionKind
+}
+
+type observerGate struct {
+	mu       sync.Mutex
+	cond     *sync.Cond
+	active   bool
+	inFlight uint32
+}
+
+func newObserverGate() *observerGate {
+	g := &observerGate{active: true}
+	g.cond = sync.NewCond(&g.mu)
+	return g
+}
+
+func (g *observerGate) begin() bool {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	if !g.active {
+		return false
+	}
+	g.inFlight++
+	return true
+}
+
+func (g *observerGate) end() {
+	g.mu.Lock()
+	g.inFlight--
+	if g.inFlight == 0 {
+		g.cond.Broadcast()
+	}
+	g.mu.Unlock()
+}
+
+func (g *observerGate) stop() {
+	g.mu.Lock()
+	g.active = false
+	for g.inFlight != 0 {
+		g.cond.Wait()
+	}
+	g.mu.Unlock()
+}
+
+type messageObserver struct {
+	id   uint64
+	gate *observerGate
+	fn   func(*MessageContext) error
+}
+
+func (o *messageObserver) invoke(ctx *MessageContext) (err error) {
+	if !o.gate.begin() {
+		return nil
+	}
+	defer o.gate.end()
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			err = fmt.Errorf("workers: message observer %d panicked: %v", o.id, recovered)
+		}
+	}()
+	return o.fn(ctx)
+}
+
+type exitObserver struct {
+	id   uint64
+	gate *observerGate
+	fn   func(*WorkerExitContext)
+}
+
+func (o *exitObserver) invoke(ctx *WorkerExitContext) (panicErr error) {
+	if !o.gate.begin() {
+		return nil
+	}
+	defer o.gate.end()
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			panicErr = fmt.Errorf("workers: exit observer %d panicked: %v", o.id, recovered)
+		}
+	}()
+	o.fn(ctx)
+	return nil
+}
+
+func (w *Workers) nextObserverIDLocked() (uint64, error) {
+	w.nextObs++
+	if w.nextObs == 0 {
+		return 0, fmt.Errorf("workers: observer ID space exhausted")
+	}
+	return w.nextObs, nil
+}
+
+// ObserveMessages registers a message observer until it is unsubscribed.
+func (w *Workers) ObserveMessages(fn func(*MessageContext) error) (Subscription, error) {
+	if w == nil || fn == nil {
+		return Subscription{}, fmt.Errorf("workers: invalid message observer")
+	}
 	w.mu.Lock()
-	defer w.mu.Unlock()
-	if !w.closed {
-		w.messages = append(w.messages, fns...)
+	if w.closed {
+		w.mu.Unlock()
+		return Subscription{}, ErrWorkerRuntimeClosed
+	}
+	id, err := w.nextObserverIDLocked()
+	if err != nil {
+		w.mu.Unlock()
+		return Subscription{}, err
 	}
+	observer := &messageObserver{id: id, gate: newObserverGate(), fn: fn}
+	w.messages[id] = observer
+	w.mu.Unlock()
+	return Subscription{id: id, kind: messageSubscription}, nil
 }
-func (w *Workers) OnExit(fns ...func(*WorkerExitContext)) {
+
+// ObserveExits registers an exit observer until it is unsubscribed.
+func (w *Workers) ObserveExits(fn func(*WorkerExitContext)) (Subscription, error) {
+	if w == nil || fn == nil {
+		return Subscription{}, fmt.Errorf("workers: invalid exit observer")
+	}
 	w.mu.Lock()
-	defer w.mu.Unlock()
-	if !w.closed {
-		w.exits = append(w.exits, fns...)
+	if w.closed {
+		w.mu.Unlock()
+		return Subscription{}, ErrWorkerRuntimeClosed
 	}
+	id, err := w.nextObserverIDLocked()
+	if err != nil {
+		w.mu.Unlock()
+		return Subscription{}, err
+	}
+	observer := &exitObserver{id: id, gate: newObserverGate(), fn: fn}
+	w.exits[id] = observer
+	w.mu.Unlock()
+	return Subscription{id: id, kind: exitSubscription}, nil
+}
+
+// Unsubscribe removes one observer and waits for callbacks already in flight.
+// It is idempotent. Do not call it from inside that observer's callback.
+func (w *Workers) Unsubscribe(subscription Subscription) error {
+	if w == nil || subscription.id == 0 {
+		return fmt.Errorf("workers: invalid subscription")
+	}
+	var gate *observerGate
+	w.mu.Lock()
+	switch subscription.kind {
+	case messageSubscription:
+		if observer := w.messages[subscription.id]; observer != nil {
+			delete(w.messages, subscription.id)
+			gate = observer.gate
+		}
+	case exitSubscription:
+		if observer := w.exits[subscription.id]; observer != nil {
+			delete(w.exits, subscription.id)
+			gate = observer.gate
+		}
+	default:
+		w.mu.Unlock()
+		return fmt.Errorf("workers: invalid subscription")
+	}
+	w.mu.Unlock()
+	if gate != nil {
+		gate.stop()
+	}
+	return nil
+}
+
+func (w *Workers) messageObservers() []*messageObserver {
+	w.mu.Lock()
+	ids := make([]uint64, 0, len(w.messages))
+	for id := range w.messages {
+		ids = append(ids, id)
+	}
+	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+	observers := make([]*messageObserver, 0, len(ids))
+	for _, id := range ids {
+		observers = append(observers, w.messages[id])
+	}
+	w.mu.Unlock()
+	return observers
+}
+
+func (w *Workers) exitObservers() []*exitObserver {
+	w.mu.Lock()
+	ids := make([]uint64, 0, len(w.exits))
+	for id := range w.exits {
+		ids = append(ids, id)
+	}
+	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+	observers := make([]*exitObserver, 0, len(ids))
+	for _, id := range ids {
+		observers = append(observers, w.exits[id])
+	}
+	w.mu.Unlock()
+	return observers
 }
 
 func normalizeOptions(o WorkerOptions) (WorkerOptions, error) {
@@ -252,7 +565,7 @@ func (w *Workers) Spawn(caller wago.HostModule, tableIndex uint32, opts WorkerOp
 	if w == nil || w.manager == nil {
 		return 0, ErrWorkersInactive
 	}
-	parent, err := w.manager.Caller(caller)
+	parent, err := w.manager.CallerIdentity(caller)
 	if err != nil {
 		return 0, ErrInvalidWorkerCaller
 	}
@@ -299,10 +612,11 @@ func (w *Workers) Spawn(caller wago.HostModule, tableIndex uint32, opts WorkerOp
 	} else {
 		w.next++
 	}
-	wr := &worker{owner: w, id: id, creator: parent, instance: child, tableIndex: tableIndex,
+	childIdentity := child.Identity()
+	wr := &worker{owner: w, id: id, creator: parent, identity: childIdentity, instance: child, tableIndex: tableIndex,
 		queue: make([]message, opts.QueueCapacity), maxPayload: opts.MaxPayloadBytes, maxQueueBytes: opts.MaxQueueBytes,
-		wake: make(chan struct{}, 1), done: make(chan struct{}), messages: append([]func(*MessageContext) error(nil), w.messages...), exits: append([]func(*WorkerExitContext){}, w.exits...)}
-	w.workers[id], w.byInstance[child.Instance()] = wr, wr
+		wake: make(chan struct{}, 1), done: make(chan struct{})}
+	w.workers[id], w.byInstance[childIdentity] = wr, wr
 	w.mu.Unlock()
 	go wr.run()
 	return id, nil
@@ -322,7 +636,7 @@ func (w *Workers) Current(caller wago.HostModule) (WorkerID, error) {
 		return 0, ErrWorkerNotFound
 	}
 	w.mu.Lock()
-	wr := w.byInstance[owned.Instance()]
+	wr := w.byInstance[owned.Identity()]
 	w.mu.Unlock()
 	if wr == nil {
 		return 0, ErrWorkerNotFound
@@ -330,22 +644,25 @@ func (w *Workers) Current(caller wago.HostModule) (WorkerID, error) {
 	return wr.id, nil
 }
 
-func (w *Workers) DispatchNext(caller wago.HostModule) error {
+func (w *Workers) DispatchNext(ctx context.Context, caller wago.HostModule) error {
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	owned, err := w.manager.ManagedCaller(caller)
 	if err != nil {
 		return ErrInvalidWorkerCaller
 	}
 	w.mu.Lock()
-	wr := w.byInstance[owned.Instance()]
+	wr := w.byInstance[owned.Identity()]
 	w.mu.Unlock()
 	if wr == nil {
 		return ErrWorkerNotFound
 	}
-	return wr.dispatch(caller)
+	return wr.dispatch(ctx, caller)
 }
 
 func (w *Workers) Link(caller wago.HostModule, childID WorkerID) error {
-	parent, err := w.manager.Caller(caller)
+	parent, err := w.manager.CallerIdentity(caller)
 	if err != nil {
 		return ErrInvalidWorkerCaller
 	}
@@ -353,7 +670,7 @@ func (w *Workers) Link(caller wago.HostModule, childID WorkerID) error {
 	if err != nil {
 		return err
 	}
-	if wr.creator != parent || wr.instance.Instance() == parent {
+	if wr.creator != parent || wr.identity == parent {
 		return ErrInvalidWorkerLink
 	}
 	wr.mu.Lock()
@@ -398,7 +715,8 @@ type worker struct {
 	mu                            sync.Mutex
 	owner                         *Workers
 	id                            WorkerID
-	creator                       *wago.Instance
+	creator                       wago.InstanceIdentity
+	identity                      wago.InstanceIdentity
 	instance                      *wago.ManagedInstance
 	tableIndex                    uint32
 	queue                         []message
@@ -409,8 +727,6 @@ type worker struct {
 	done                          chan struct{}
 	stopping, dispatching, linked bool
 	stopErr                       error
-	messages                      []func(*MessageContext) error
-	exits                         []func(*WorkerExitContext)
 }
 
 func (wr *worker) signal() {
@@ -443,7 +759,7 @@ func (wr *worker) enqueue(tag uint64, payload []byte) error {
 	return nil
 }
 
-func (wr *worker) dispatch(caller wago.HostModule) error {
+func (wr *worker) dispatch(ctx context.Context, caller wago.HostModule) error {
 	wr.mu.Lock()
 	if wr.dispatching {
 		wr.mu.Unlock()
@@ -472,8 +788,8 @@ func (wr *worker) dispatch(caller wago.HostModule) error {
 			wr.queuedBytes -= uint32(len(msg.payload))
 			wr.mu.Unlock()
 			ctx := &MessageContext{WorkerID: wr.id, Tag: msg.tag, Payload: msg.payload, Caller: caller}
-			for _, fn := range wr.messages {
-				if err := fn(ctx); err != nil {
+			for _, observer := range wr.owner.messageObservers() {
+				if err := observer.invoke(ctx); err != nil {
 					wr.stop(err)
 					return err
 				}
@@ -485,6 +801,8 @@ func (wr *worker) dispatch(caller wago.HostModule) error {
 		case <-wr.wake:
 		case <-expired:
 			return ErrInvalidWorkerCaller
+		case <-ctx.Done():
+			return ctx.Err()
 		}
 	}
 }
@@ -515,24 +833,18 @@ func (wr *worker) run() {
 	} else if err != nil {
 		kind = WorkerFailed
 	}
-	in := wr.instance.Instance()
 	_ = wr.instance.Close()
 	wr.owner.mu.Lock()
 	delete(wr.owner.workers, wr.id)
-	delete(wr.owner.byInstance, in)
+	delete(wr.owner.byInstance, wr.identity)
 	wr.owner.mu.Unlock()
 	ctx := &WorkerExitContext{WorkerID: wr.id, Kind: kind, Err: err}
-	for i, fn := range wr.exits {
-		func() {
-			defer func() {
-				if v := recover(); v != nil {
-					wr.owner.mu.Lock()
-					wr.owner.exitPanics = append(wr.owner.exitPanics, fmt.Errorf("worker %d exit observer %d: %v", wr.id, i, v))
-					wr.owner.mu.Unlock()
-				}
-			}()
-			fn(ctx)
-		}()
+	for _, observer := range wr.owner.exitObservers() {
+		if panicErr := observer.invoke(ctx); panicErr != nil {
+			wr.owner.mu.Lock()
+			wr.owner.exitPanics = append(wr.owner.exitPanics, fmt.Errorf("worker %d: %w", wr.id, panicErr))
+			wr.owner.mu.Unlock()
+		}
 	}
 	// Release aggregate quota only after the instance is closed and every exit
 	// observer has run, so a concurrent Spawn cannot exceed the ceiling while this
@@ -541,7 +853,7 @@ func (wr *worker) run() {
 	close(wr.done)
 }
 
-func (w *Workers) parentClosing(parent *wago.Instance) {
+func (w *Workers) parentClosing(parent wago.InstanceIdentity) {
 	w.mu.Lock()
 	var linked []*worker
 	for _, wr := range w.workers {
@@ -578,8 +890,20 @@ func (w *Workers) close() error {
 		<-wr.done
 	}
 	w.mu.Lock()
+	observers := make([]*observerGate, 0, len(w.messages)+len(w.exits))
+	for _, observer := range w.messages {
+		observers = append(observers, observer.gate)
+	}
+	for _, observer := range w.exits {
+		observers = append(observers, observer.gate)
+	}
+	w.messages = nil
+	w.exits = nil
 	errs := append([]error(nil), w.exitPanics...)
 	w.exitPanics = nil
 	w.mu.Unlock()
+	for _, observer := range observers {
+		observer.stop()
+	}
 	return errors.Join(errs...)
 }
diff --git a/workers_test.go b/workers_test.go
index cded273..c8a70d7 100644
--- a/workers_test.go
+++ b/workers_test.go
@@ -1,43 +1,128 @@
 package workers
 
 import (
+	"context"
+	"encoding/json"
 	"errors"
+	"sort"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/wago-org/wago"
-	"github.com/wago-org/wago/testutil/wasmtest"
+	wagoplugin "github.com/wago-org/wago/plugin"
+	"github.com/wago-org/wago/tests/wasmtest"
 )
 
+type registerFunc func(*wago.Registrar) error
+
+func (f registerFunc) Register(reg *wago.Registrar) error { return f(reg) }
+
+func pluginTestSet(t *testing.T, providers []wago.PluginProvider, config json.RawMessage) wago.PluginSet {
+	t.Helper()
+	set := wago.PluginSet{Providers: providers}
+	for _, provider := range providers {
+		digest, err := wago.DefinitionDigest(provider.Definition)
+		if err != nil {
+			t.Fatal(err)
+		}
+		selection := wago.PluginSelection{
+			ID: provider.Definition.ID, DefinitionDigest: digest, Direct: true,
+			Dependencies: map[string]string{},
+		}
+		for _, requirement := range provider.Definition.Requires {
+			selection.Dependencies[requirement.ID] = requirement.Version
+		}
+		if provider.Definition.ID == PluginID {
+			selection.Config = append(json.RawMessage(nil), config...)
+		}
+		for _, authority := range provider.Definition.Authorities {
+			selection.Grants = append(selection.Grants, wago.AuthorityGrant{Name: authority.Name, Scope: authority.Scope})
+		}
+		for _, requirement := range provider.Definition.Consumes {
+			var owners []string
+			for _, candidate := range providers {
+				for _, provided := range candidate.Definition.Provides {
+					if provided.ID == requirement.ID && provided.Major == requirement.Major {
+						owners = append(owners, candidate.Definition.ID)
+					}
+				}
+			}
+			sort.Strings(owners)
+			if requirement.Mode != wago.ContractMany && len(owners) > 1 {
+				owners = owners[:1]
+			}
+			selection.Contracts = append(selection.Contracts, wago.ContractBinding{ID: requirement.ID, Major: requirement.Major, Providers: owners})
+		}
+		set.Selections = append(set.Selections, selection)
+	}
+	return set
+}
+
 type integrationPlugin struct {
-	Plugin
+	service  *wagoplugin.Ref[Service]
 	id       WorkerID
 	exits    chan WorkerExitContext
 	messages chan MessageContext
+	message  Subscription
+	exit     Subscription
 }
 
-func (p *integrationPlugin) Info() wago.ExtensionInfo {
-	info := p.Plugin.Info()
-	info.ID = "wago.workers.integration"
-	info.RequiresCapabilities = append(info.RequiresCapabilities, wago.PluginHostImports)
-	return info
-}
-
-func (p *integrationPlugin) Register(reg *wago.Registry) error {
-	if err := p.Plugin.Register(reg); err != nil {
-		return err
-	}
-	p.Service().OnMessage(func(ctx *MessageContext) error { p.messages <- *ctx; return nil })
-	p.Service().OnExit(func(ctx *WorkerExitContext) { p.exits <- *ctx })
-	imports, err := reg.HostImports()
-	if err != nil {
-		return err
+func integrationProvider(p *integrationPlugin) wago.PluginProvider {
+	definition := wago.PluginDefinition{
+		ID: "example.com/workers-integration", Version: "1.0.0",
+		Provenance: wago.PluginProvenance{Repository: "https://example.com/workers-integration", License: "MIT"},
+		Requires:   []wago.PluginRequirement{{ID: PluginID, Version: "^0.1.0"}},
+		Authorities: []wago.AuthorityRequest{{
+			Name: wago.AuthorityHostImportDefine, Mode: wago.AuthorityRequired,
+			Reason: "exercise worker calls from a guest host import",
+			Scope:  wago.AuthorityScope{Modules: []string{"test"}},
+		}},
+		Consumes: []wago.ContractRequirement{{ID: Contract.ID(), Major: Contract.Major(), Mode: wago.ContractRequired}},
 	}
-	imports.Module("test").Func("spawn", func(caller wago.HostModule, _, _ []uint64) {
-		p.id, _ = p.Service().Spawn(caller, 0, WorkerOptions{QueueCapacity: 1, MaxPayloadBytes: 16, MaxQueueBytes: 16})
-	})
-	imports.Module("test").Func("next", func(caller wago.HostModule, _, _ []uint64) { _ = p.Service().DispatchNext(caller) })
-	return nil
+	return wago.PluginProvider{Definition: definition, New: func() wago.Plugin {
+		return registerFunc(func(reg *wago.Registrar) error {
+			var err error
+			p.service, err = wagoplugin.Require(reg, Contract)
+			if err != nil {
+				return err
+			}
+			imports, err := reg.HostImports()
+			if err != nil {
+				return err
+			}
+			module, err := imports.Module("test")
+			if err != nil {
+				return err
+			}
+			module.Func("spawn", func(caller wago.HostModule, _, _ []uint64) {
+				_ = p.service.With(func(service Service) error {
+					p.id, _ = service.Spawn(caller, 0, WorkerOptions{QueueCapacity: 1, MaxPayloadBytes: 16, MaxQueueBytes: 16})
+					return nil
+				})
+			})
+			module.Func("next", func(caller wago.HostModule, _, _ []uint64) {
+				_ = p.service.With(func(service Service) error { return service.DispatchNext(context.Background(), caller) })
+			})
+			return reg.Lifecycle(wago.PluginLifecycle{
+				Start: func(context.Context) error {
+					return p.service.With(func(service Service) error {
+						p.message, err = service.ObserveMessages(func(ctx *MessageContext) error { p.messages <- *ctx; return nil })
+						if err != nil {
+							return err
+						}
+						p.exit, err = service.ObserveExits(func(ctx *WorkerExitContext) { p.exits <- *ctx })
+						return err
+					})
+				},
+				Stop: func(context.Context) error {
+					return p.service.With(func(service Service) error {
+						return errors.Join(service.Unsubscribe(p.message), service.Unsubscribe(p.exit))
+					})
+				},
+			})
+		})
+	}}
 }
 
 func integrationModule() []byte {
@@ -61,53 +146,32 @@ func integrationModule() []byte {
 }
 
 func TestWorkerLimitsQuota(t *testing.T) {
-	// Defaults apply when a field is zero.
 	if got := normalizeLimits(WorkerLimits{}); got.MaxLiveWorkers != DefaultMaxLiveWorkers || got.MaxQueueBytes != DefaultMaxServiceQueueBytes {
 		t.Fatalf("normalizeLimits(zero) = %+v", got)
 	}
-	if got := normalizeLimits(WorkerLimits{MaxLiveWorkers: 3}); got.MaxQueueBytes != DefaultMaxServiceQueueBytes || got.MaxLiveWorkers != 3 {
-		t.Fatalf("normalizeLimits partial = %+v", got)
-	}
-
-	// MaxLiveWorkers ceiling: the third reservation is rejected, and a release frees a slot.
-	w := newWorkers(nil, WorkerLimits{MaxLiveWorkers: 2, MaxQueueBytes: 1 << 20})
-	if err := w.reserve(100); err != nil {
-		t.Fatalf("reserve 1: %v", err)
-	}
-	if err := w.reserve(100); err != nil {
-		t.Fatalf("reserve 2: %v", err)
-	}
-	if err := w.reserve(100); !errors.Is(err, ErrWorkerQuotaExceeded) {
-		t.Fatalf("reserve 3 = %v, want ErrWorkerQuotaExceeded", err)
-	}
-	w.release(100)
-	if err := w.reserve(100); err != nil {
-		t.Fatalf("reserve after release: %v", err)
-	}
-
-	// Aggregate queue-byte ceiling is enforced independently and cannot overflow.
-	w2 := newWorkers(nil, WorkerLimits{MaxLiveWorkers: 100, MaxQueueBytes: 1000})
-	if err := w2.reserve(600); err != nil {
-		t.Fatalf("reserve 600: %v", err)
+	w := newWorkers(nil, WorkerLimits{MaxLiveWorkers: 2, MaxQueueBytes: 1000})
+	if err := w.reserve(600); err != nil {
+		t.Fatal(err)
 	}
-	if err := w2.reserve(600); !errors.Is(err, ErrWorkerQuotaExceeded) {
-		t.Fatalf("reserve 600 over budget = %v, want ErrWorkerQuotaExceeded", err)
+	if err := w.reserve(400); err != nil {
+		t.Fatal(err)
 	}
-	if err := w2.reserve(400); err != nil {
-		t.Fatalf("reserve exact remaining 400: %v", err)
+	if err := w.reserve(1); !errors.Is(err, ErrWorkerQuotaExceeded) {
+		t.Fatalf("reserve above quota = %v", err)
 	}
-
-	// A closed service rejects reservations.
-	w2.closed = true
-	if err := w2.reserve(1); !errors.Is(err, ErrWorkerRuntimeClosed) {
-		t.Fatalf("reserve on closed = %v, want ErrWorkerRuntimeClosed", err)
+	w.release(600)
+	if err := w.reserve(1); err != nil {
+		t.Fatal(err)
 	}
 }
 
-func TestPluginSpawnsCopiesMessageAndStops(t *testing.T) {
+func TestPluginContractSpawnsCopiesMessageAndStops(t *testing.T) {
 	p := &integrationPlugin{exits: make(chan WorkerExitContext, 1), messages: make(chan MessageContext, 1)}
 	rt := wago.NewRuntime()
-	if err := rt.Use(p); err != nil {
+	// Reverse the dependency order to prove Requires plus the reviewed contract
+	// binding, rather than fixture order, places Workers before its consumer.
+	set := pluginTestSet(t, []wago.PluginProvider{integrationProvider(p), Provider()}, json.RawMessage(`{"maxLiveWorkers":8,"maxQueueBytes":1024}`))
+	if err := rt.LoadPlugins(context.Background(), set); err != nil {
 		t.Fatal(err)
 	}
 	mod, err := rt.Compile(integrationModule())
@@ -126,7 +190,7 @@ func TestPluginSpawnsCopiesMessageAndStops(t *testing.T) {
 		t.Fatal("spawn returned zero ID")
 	}
 	payload := []byte("abc")
-	if err := p.Service().Send(p.id, 42, payload); err != nil {
+	if err := p.service.With(func(service Service) error { return service.Send(p.id, 42, payload) }); err != nil {
 		t.Fatal(err)
 	}
 	copy(payload, "zzz")
@@ -146,10 +210,62 @@ func TestPluginSpawnsCopiesMessageAndStops(t *testing.T) {
 	case <-time.After(2 * time.Second):
 		t.Fatal("exit timeout")
 	}
-	if err := p.Service().Send(p.id, 0, nil); !errors.Is(err, ErrWorkerNotFound) {
-		t.Fatalf("send after exit = %v", err)
-	}
 	if err := rt.Close(); err != nil {
 		t.Fatal(err)
 	}
+	if err := p.service.With(func(Service) error { return nil }); !errors.Is(err, wago.ErrPermissionDenied) {
+		t.Fatalf("contract after close = %v", err)
+	}
+}
+
+func TestSubscriptionCloseWaitsAndPreventsFutureCalls(t *testing.T) {
+	w := newWorkers(nil, WorkerLimits{})
+	started, release := make(chan struct{}), make(chan struct{})
+	var calls atomic.Int32
+	sub, err := w.ObserveMessages(func(*MessageContext) error {
+		calls.Add(1)
+		close(started)
+		<-release
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	observer := w.messageObservers()[0]
+	done := make(chan struct{})
+	go func() { _ = observer.invoke(new(MessageContext)); close(done) }()
+	<-started
+	closed := make(chan struct{})
+	go func() { _ = w.Unsubscribe(sub); close(closed) }()
+	select {
+	case <-closed:
+		t.Fatal("subscription closed before callback returned")
+	case <-time.After(20 * time.Millisecond):
+	}
+	close(release)
+	<-done
+	<-closed
+	if err := observer.invoke(new(MessageContext)); err != nil {
+		t.Fatal(err)
+	}
+	if got := calls.Load(); got != 1 {
+		t.Fatalf("observer calls = %d, want 1", got)
+	}
+}
+
+func TestPluginRejectsStrictInvalidConfig(t *testing.T) {
+	for _, config := range []json.RawMessage{
+		json.RawMessage(`null`),
+		json.RawMessage(`[]`),
+		json.RawMessage(`{"maxLiveWorkers":null}`),
+		json.RawMessage(`{"maxLiveWorkers":1,"maxLiveWorkers":2}`),
+		json.RawMessage(`{"unknown":1}`),
+		json.RawMessage(`{"maxLiveWorkers":0}`),
+		json.RawMessage(`{"maxQueueBytes":0}`),
+		json.RawMessage(`{} {}`),
+	} {
+		if err := wago.ValidatePluginSet(pluginTestSet(t, []wago.PluginProvider{Provider()}, config)); err == nil {
+			t.Fatalf("accepted invalid config %s", config)
+		}
+	}
 }