feat: actor deregistration with safety checks, CLI, admin UI (#56)

CHANGELOG.md

@@ -203,6 +203,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `stranded_jobs_query_failed`, and the `failed_details` payload of
   `sub_enqueue_flush_failed`). Log pipelines querying `fields.message`
   on these two events must switch to `error_message`.
+- **Breaking: `taskq.worker.actor_config` moved to `taskq.actor_config`.**
+  The `ActorConfig` dataclass (released in v0.2.0–v0.2.2 at
+  `taskq.worker.actor_config`) has moved to the top-level
+  `taskq.actor_config` module. It is shared by the client, CLI, and admin
+  UI, not worker-internal. The old import path raises `ImportError`. See
+  [docs/guides/upgrading.md](docs/guides/upgrading.md) for the full
+  migration mapping. The companion `actor_config_ops` module (listing,
+  inspecting, tuning, and deregistering actors) has likewise moved from
+  `taskq.worker.actor_config_ops` to `taskq.actor_config_ops`; it was
+  never released under the `worker.*` path.
 
 ### Security
 

docs/api-reference/client.md

@@ -1,9 +1,11 @@
 # Client
 
-`TaskQ`, `JobsClient`, `JobHandle`, and `CancelResult`.
+`TaskQ`, `JobsClient`, `JobHandle`, `CancelResult`, and `ActorsClient`.
 
 ::: taskq.client._taskq
 
 ::: taskq.client._jobs
 
 ::: taskq.client._handle
+
+::: taskq.client._actors.ActorsClient

docs/guides/actors.md

@@ -25,6 +25,7 @@ the actor's payload and result types end-to-end.
 15. [Progress reporting](#progress-reporting)
 16. [Testing actors without a database](#testing-actors-without-a-database)
 17. [Full worked example](#full-worked-example)
+18. [Actor deregistration](#actor-deregistration)
 
 ---
 
@@ -944,3 +945,82 @@ async def submit_order(client, order_id: str, customer_id: str, amount_cents: in
     print(f"confirmed: {result.confirmation_number}")
     return handle.job_id
 ```
+
+---
+
+## Actor deregistration
+
+Actors registered by worker startup create `actor_config` rows that persist
+until explicitly removed. For long-lived deployments this is intentional —
+the row is the source of truth for capacity and routing. For ephemeral,
+per-run deployments (e.g. `my-actor.<run-id>`), each run leaves a row behind.
+
+### `client.actors.deregister()`
+
+```python
+async with TaskQ(dsn=...) as tq:
+    result = await tq.actors.deregister("my-actor.run-123")
+    # force=False: refuses if non-terminal jobs or enabled schedules exist
+```
+
+**Safety checks (force=False):**
+- Refuses if any non-terminal jobs (pending/scheduled/running) reference the
+  actor.
+- Refuses if any enabled cron schedules reference the actor.
+
+**force=True:**
+- Still refuses if **running** jobs exist (they are actively executing).
+- Cancels pending/scheduled jobs (marks as `cancelled` with
+  `error_class='ActorDeregistered'`).
+- Disables enabled cron schedules (sets `enabled=false`).
+
+**Terminal job history** is never deleted. The `jobs.actor` column is plain
+text, not a foreign key — terminal rows remain queryable by actor name after
+deregistration.
+
+**Queue cleanup** (`purge_queue=True`): deletes the `queues` row if no other
+`actor_config` references the same queue. A shared queue is never purged.
+
+### Enqueue after deregistration
+
+After deregistration, any client can still `enqueue()` the dead actor name —
+the `INSERT` succeeds (there is no foreign key from `jobs.actor` to
+`actor_config.actor`), and the job sits in `pending` status forever. Because
+the dispatch query inner-joins `actor_config`, the job will **never be
+dispatched** and no background sweep will reap it.
+
+**Operational discipline:** stop enqueuing to an actor *before* deregistering
+it. Deregistration is best-effort against concurrent enqueue/dispatch;
+callers must quiesce the actor first.
+
+**Stop workers first:** A concurrent worker startup (`sync_actor_config`)
+can re-create the `actor_config` row after deregistration, with capacity
+fields reset to `@actor(...)` defaults. Stop all workers for the actor
+before calling `deregister`.
+
+### Idempotent deregistration
+
+A second `deregister` call on an already-deregistered actor raises
+`ActorNotFoundError`. For cleanup-automation loops:
+
+```python
+from taskq.exceptions import ActorNotFoundError
+
+try:
+    await tq.actors.deregister(actor_name, force=True, purge_queue=True)
+except ActorNotFoundError:
+    pass  # already deregistered — idempotent
+```
+
+### CLI
+
+```bash
+taskq actor-config deregister my-actor.run-123
+taskq actor-config deregister my-actor.run-123 --force --purge-queue
+```
+
+### Admin UI
+
+The `/admin/actors` page lists all `actor_config` rows with active job counts
+and schedule counts. Each row has a deregister form with `force` and
+`purge_queue` checkboxes (requires `TASKQ_ADMIN_ACTIONS_ENABLED=true`).

docs/guides/admin-ui.md

@@ -342,6 +342,32 @@ Rate-limit state page. Reads all rows from `rate_limit_buckets` (bucket name, ki
 
 Reservation slot summary. For each `bucket_name` in `reservation_slots`, shows the count of held slots (where `job_id IS NOT NULL`), free slots, and total slots.
 
+### `GET /admin/actors`
+
+The `/admin/actors` page lists all stored `actor_config` rows with:
+
+- Actor name, queue, max concurrent, max pending
+- Active job count (pending + scheduled + running)
+- Enabled schedule count
+- Last updated timestamp
+
+Each row has a **Deregister** button with `force` and `purge queue` checkboxes.
+Deregistration requires `TASKQ_ADMIN_ACTIONS_ENABLED=true`. The form is
+CSRF-protected via the synchronizer-token pattern.
+
+### POST /admin/actors/{actor}/deregister
+
+Deregisters an actor. Form fields:
+- `csrf_token` — CSRF synchronizer token (set by GET)
+- `force` — checkbox; cancels pending/scheduled jobs and disables schedules
+- `purge_queue` — checkbox; deletes the orphaned queues row
+
+Response codes:
+- `303` — success, redirects to `/actors?notice=deregistered+{actor}`
+- `403` — admin actions disabled or CSRF validation failed
+- `404` — actor not found (no `actor_config` row)
+- `409` — actor has active jobs or enabled schedules (force=False)
+
 ### `GET /admin/sse/{topic}`
 
 SSE (Server-Sent Events) endpoint. Accepts any `topic` string. On connect it emits an initial `event: status` frame with `{"status": "awaiting_progress_backend"}`, then sends `: keepalive` comments every 30 seconds to prevent connection timeout. See [Real-time vs polling mode](#real-time-vs-polling-mode) below.

docs/guides/cli.md

@@ -303,6 +303,22 @@ Per actor and capacity field this prints the `@actor(...)` literal, the stored v
 
 `taskq actor-config set` requires the actor to have a stored row already (created by a worker startup that registered it). `queue` and `metadata` are structural and are only ever changed by redeploying with a new `@actor(...)` registration (plus `--force-update-actor-config` if a stored row already exists).
 
+### `taskq actor-config deregister`
+
+Deregister an actor: delete its `actor_config` row with safety checks.
+
+```bash
+taskq actor-config deregister <ACTOR> [--force] [--purge-queue]
+```
+
+- `<ACTOR>` — actor name (positional argument)
+- `--force` — cancel pending/scheduled jobs, disable enabled cron schedules,
+  and proceed despite non-terminal jobs. Running jobs still block.
+- `--purge-queue` — also delete the orphaned `queues` row if no other actor
+  references it.
+
+Exit code 0 on success, 1 on refusal (with error message) or not found.
+
 ### Exit codes
 
 | Code | Meaning |

docs/guides/jobs-clients.md

@@ -4,6 +4,9 @@
 `Backend` and adds typed payload serialisation, `JobHandle[R]` construction, and
 `CancelResult` building.
 
+**Actor management:** `tq.actors` provides `list()`, `get()`, `set_capacity()`,
+and `deregister()` — see [Actor deregistration](actors.md#actor-deregistration).
+
 ---
 
 ## Job lifecycle

docs/guides/upgrading.md

@@ -61,3 +61,61 @@ This is a deliberate tradeoff, not a missing feature:
 - Pin `taskq-py` back to the previous version until the issue is resolved,
   since the previous version's code may not be compatible with the new
   schema.
+
+---
+
+## Breaking import path changes
+
+### `taskq.worker.actor_config` → `taskq.actor_config`
+
+> **Released in v0.2.0–v0.2.2, moved in unreleased.** This is a breaking
+> change for anyone importing `ActorConfig` from the old path.
+
+The `ActorConfig` dataclass has moved from `taskq.worker.actor_config` to
+the top-level `taskq.actor_config` module. It is a shared carrier used by
+the client, CLI, and admin UI — not worker-internal.
+
+**Old (v0.2.0–v0.2.2):**
+
+```python
+from taskq.worker.actor_config import ActorConfig
+```
+
+**New:**
+
+```python
+from taskq.actor_config import ActorConfig
+```
+
+The old import path raises `ImportError` — update your imports.
+
+### `taskq.worker.actor_config_ops` → `taskq.actor_config_ops`
+
+The `actor_config_ops` module — listing, inspecting, tuning, and
+deregistering actors on a live deployment — has moved from
+`taskq.worker.actor_config_ops` to the top-level
+`taskq.actor_config_ops`. This module was introduced on the unreleased
+branch; if you were importing it from the `worker.*` path during
+development, update to the top-level path.
+
+**Old (unreleased branch only):**
+
+```python
+from taskq.worker.actor_config_ops import (
+    list_actor_configs,
+    get_actor_config,
+    set_actor_config_capacity,
+    deregister_actor,
+)
+```
+
+**New:**
+
+```python
+from taskq.actor_config_ops import (
+    list_actor_configs,
+    get_actor_config,
+    set_actor_config_capacity,
+    deregister_actor,
+)
+```

pyproject.toml

@@ -274,6 +274,9 @@ ignore = [
 # Same rationale — schema name validated against _IDENT_RE; all
 # user-supplied values use $N parameter binding.
 "tests/test_web_admin_integration.py" = ["S608"]
+# Same rationale — schema name validated against _IDENT_RE; all
+# user-supplied values use $N parameter binding.
+"tests/test_web_admin_actors.py" = ["S608"]
 # FastAPI Depends() in test route-handler defaults is the same idiomatic
 # declarative-injection pattern as web/admin — not a real mutable default.
 "tests/test_sso_session.py" = ["B008"]

src/taskq/__init__.py

@@ -11,6 +11,7 @@
 import importlib.metadata
 
 from taskq.actor import ActorFn, ActorFnWithCtx, ActorHandler, ActorRef, actor
+from taskq.actor_config_ops import ActorConfigRow, DeregisterResult
 from taskq.auth import (
     PgCredential,
     PgCredentialProvider,
@@ -38,13 +39,18 @@
 )
 from taskq.batch import BatchCompletionStatus, BatchHandle, EnqueueItem, wait_for_batch
 from taskq.client import CancelResult, JobEvent, JobHandle, JobsClient, TaskQ
+from taskq.client._actors import ActorsClient
 from taskq.client._enqueuer import SubJobEnqueuer
 from taskq.connections import ConnFactory, PoolFactory, RedisFactory, WorkerConnections
 from taskq.context import JobContext
 from taskq.cron import CronScheduleSpec, ScheduleHandle, cron
 from taskq.exceptions import (
     ActorConfigDriftError,
     ActorConfigDriftList,
+    ActorDeregistrationError,
+    ActorHasActiveJobsError,
+    ActorHasEnabledSchedulesError,
+    ActorNotFoundError,
     BackpressureError,
     DependencyCycle,
     DIError,
@@ -86,10 +92,16 @@
 __all__ = [
     "ActorConfigDriftError",
     "ActorConfigDriftList",
+    "ActorConfigRow",
+    "ActorDeregistrationError",
     "ActorFn",
     "ActorFnWithCtx",
     "ActorHandler",
+    "ActorHasActiveJobsError",
+    "ActorHasEnabledSchedulesError",
+    "ActorNotFoundError",
     "ActorRef",
+    "ActorsClient",
     "BackpressureError",
     "BatchCompletionStatus",
     "BatchHandle",
@@ -99,6 +111,7 @@
     "CronScheduleSpec",
     "DIError",
     "DependencyCycle",
+    "DeregisterResult",
     "DstStrategy",
     "EnqueueItem",
     "ErrorReporter",