feat: add count-based work admission ceilings

docker/test/integration/backend-dind.yml

@@ -177,6 +177,8 @@ services:
         SOURCE_HASH: ${WORKER_SOURCE_HASH:-}
     # Uses default entrypoint.sh which runs alembic migrations before uvicorn
     environment:
+      - ENVIRONMENT=test
+      - SEED_SYSTEM_CONFIGS_ON_START=true
       - DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/postgres
       - REDIS_URL=redis://redis:6379/0
       - SECRETS_ENCRYPTION_KEY=wHhIQWmPfLt60oHdxzbQhY1ZKnUon12e5_SuZ33xDxc=

docker/test/integration/backend.yml

@@ -81,6 +81,8 @@ services:
       args:
         SOURCE_HASH: ${WORKER_SOURCE_HASH:-}
     environment:
+      - ENVIRONMENT=test
+      - SEED_SYSTEM_CONFIGS_ON_START=true
       - DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/postgres
       - REDIS_URL=redis://redis:6379/0
       - SECRETS_ENCRYPTION_KEY=wHhIQWmPfLt60oHdxzbQhY1ZKnUon12e5_SuZ33xDxc=

docker/test/integration/frontend.yml

@@ -8,6 +8,8 @@ services:
       args:
         SOURCE_HASH: ${WORKER_SOURCE_HASH:-}
     environment:
+      - ENVIRONMENT=test
+      - SEED_SYSTEM_CONFIGS_ON_START=true
       - DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/postgres
       - REDIS_URL=redis://redis:6379/0
       - LK_JWT_SECRET=test-lk-jwt-secret

docker/test/integration/infra.yml

@@ -41,9 +41,9 @@ services:
         SOURCE_HASH: ${WORKER_SOURCE_HASH:-}
     command: ["/entrypoint.sh"]
     environment:
+      - ENVIRONMENT=test
       - DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/postgres
       - REDIS_URL=redis://redis:6379/0
-      - ENVIRONMENT=test
       - LK_JWT_SECRET=test-lk-jwt-secret
       - INTERNAL_API_KEY=test-internal-api-key-for-integration-tests
     depends_on:

docker/test/integration/po-tools.yml

@@ -9,6 +9,8 @@ services:
         SOURCE_HASH: ${WORKER_SOURCE_HASH:-}
     command: ["/entrypoint.sh"]
     environment:
+      - ENVIRONMENT=test
+      - SEED_SYSTEM_CONFIGS_ON_START=true
       - DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/postgres
       - REDIS_URL=redis://redis:6379/0
       - ENVIRONMENT=test

docker/test/service/api.yml

@@ -12,6 +12,8 @@ services:
       - DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/postgres
       - REDIS_URL=redis://redis:6379/0
       - ENVIRONMENT=test
+      - SEED_SYSTEM_CONFIGS_ON_START=true
+      - SYSTEM_CONFIGS_TEST_OVERLAY=/app/scripts/system_configs.service_test.yaml
       - SECRETS_ENCRYPTION_KEY=wHhIQWmPfLt60oHdxzbQhY1ZKnUon12e5_SuZ33xDxc=
       - LK_JWT_SECRET=test-lk-jwt-secret-for-service-tests
       - INTERNAL_API_KEY=test-internal-api-key-for-service-tests

docker/test/service/scheduler.yml

@@ -31,6 +31,8 @@ services:
         SOURCE_HASH: ${WORKER_SOURCE_HASH:-}
     command: ["/entrypoint.sh"]
     environment:
+      - ENVIRONMENT=test
+      - SEED_SYSTEM_CONFIGS_ON_START=true
       - DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/postgres
       - REDIS_URL=redis://redis:6379/0
       - ENVIRONMENT=test

docs/CHANGELOG.md

@@ -5,6 +5,26 @@
 ### Added
 
 - One-time promo-code registration that atomically arms an enabled engineering budget policy.
+- Count-based work admission for projects and concurrent engineering/QA runs,
+  with an internal/admin emergency-stop API and durable typed decision audit
+  records. Paid runs now start through one transactional API command, which
+  holds the counted slot through queued-Run creation and checks engineering
+  money only after the count gate.
+
+### Fixed
+
+- Updated paid-run test fixtures to use the transactional paid-run command;
+  fixtures that only require a Run record now use a non-paid type.
+- Paid-run starts now replay an identical stable command idempotently and reject
+  conflicting payloads; the obsolete standalone paid-work admission oracle was
+  removed. Emergency-stop writes are strict booleans and admission controls are
+  protected from the generic configuration mutation API.
+- Paid-work refusals now persist their command identity, project owner, typed
+  reason and Russian owner-facing text without caching a transient outcome.
+- A paid-run retry now rechecks controls rather than caching a prior refusal;
+  it only reuses a live Run whose engineering reservation remains active.
+- Scheduler admission refusals now park their Task as well as the Story, and a
+  handled pre-handoff publish failure closes the Run with its released hold.
 
 ## 2026-08-24
 

docs/CONTRACTS.md

@@ -80,6 +80,59 @@ are hard-deleted; all accounting facts and the resolved `user_id` remain immutab
 
 ### Engineering budget policies
 
+### Count-based work admission
+
+`work_admission` is independent of engineering money. Project creation uses its
+per-user lock directly. Paid coding-agent work uses `POST /api/work-admission/paid-runs`:
+the command locks the controls, checks the emergency stop and concurrent-run
+ceiling, performs engineering's existing money admission when applicable, then
+creates the queued Run before the transaction commits. No separate successful
+paid-work admission exists. Every count decision is stored in
+`work_admission_audits` with a typed outcome and reason.
+
+Paid-run retries are decided only by that command while it holds the admission
+control locks. An existing Run is replayed only when it is `queued` or
+`running` and, for a budgeted engineering attempt, its reservation is still
+`active`. Audit rows preserve command identity for payload-conflict detection;
+they never cache a denial. A retry after a stop is lifted or a capacity slot is
+freed therefore evaluates the controls again. A terminal Run is never reopened:
+its identity returns the typed `paid_run_identity_expired` conflict and a caller
+must create a new attempt identity. A changed payload under a terminal identity
+returns `paid_run_command_conflict` before that expiry outcome.
+
+For scheduler dispatch and QA handoff, an addressable refusal writes its owed
+owner-notification record before the task/story transitions that park work;
+operator-facing spawn-worker and run-e2e instead return the typed result
+synchronously, while the command still records its audit reason.
+
+Known limitations: if a process dies after the paid-run command commits and
+before handoff, its queued Run continues to occupy the ceiling until manual
+intervention. The atomic internal abort command is used only when preparation
+failed before any queue call; it marks the Run `cancelled` and releases its
+hold. A publication exception has an unknown broker outcome, so the queued Run
+and its active hold remain for normal unfinished-run recovery rather than being
+incorrectly cancelled.
+Refusal notification is attached to the project's initiating Run, so a second
+task refusal for that project can be suppressed and a standalone task has no
+owner notification.
+
+The deployed defaults in `scripts/system_configs.yaml` are:
+
+- `work_admission.max_projects_per_user=3`, measured as non-archived projects
+  owned by one non-admin user. Deleted projects are absent and archived projects
+  do not count; administrators are unlimited.
+- `work_admission.max_concurrent_paid_runs=5`, measured globally as queued or
+  running `engineering` and `qa` runs together.
+- `work_admission.emergency_stop=false`. Internal/admin callers read and set
+  this one operator switch at `/api/work-admission/emergency-stop`; while true,
+  no new project, engineering run, or QA run is admitted. The stored value must
+  be a boolean; malformed configuration fails closed.
+  It never changes existing rows, workspaces, containers, or runs.
+
+Engineering's count check precedes its existing monetary
+`admit_engineering_attempt` inside that same command, so a count-based refusal
+cannot create a financial reservation. QA never enters the monetary gate.
+
 `engineering_budget_policies` holds at most one durable policy row per `user_id`.
 `limit_microusd` is a non-negative integer number of micro-USD; budget requests never
 accept floating-point or dollar-denominated money. `state` is the typed
@@ -120,13 +173,14 @@ denies. `POST .../admissions/{attempt_id}/release` may release only a proven pre
 `active` hold.
 
 `engineering_budget_reservations` records those decisions separately from the immutable
-ledger. The pre-handoff boundary ends only when the engineering message has published.
-Dispatchers validate cheap local conditions first; after an admitted `active` hold, every
-exception or typed refusal before that boundary, including Run creation, recipient
-resolution and publishing, changes it to `released`. A released row proves only that its
-previous handoff did not begin; a deterministic replay such as a deploy-fix dispatch must
-re-enter admission and obtain a newly `active` row before any story transition, Run creation
-or queue publication. This applies to ordinary task
+ledger. The pre-handoff boundary ends before an engineering message is submitted to the
+queue. Dispatchers validate cheap local conditions first; after an admitted `active` hold,
+a typed refusal or an exception proven to occur before that queue call — including Run
+creation and recipient resolution — changes it to `released`. An exception from publication
+has an unknown broker outcome and does not release the hold or cancel the queued Run. A
+released row proves only that its previous handoff did not begin; a deterministic replay such
+as a deploy-fix dispatch must re-enter admission and obtain a newly `active` row before any
+story transition, Run creation or queue publication. This applies to ordinary task
 dispatch and supervisor deploy-fix dispatch, whose stable attempt id is
 `eng-deploy-fix-{deploy_run_id}-{attempt}`. A scheduler denial has no Run or queue side
 effect and moves the affected task or deploy-fix story to `waiting_human_review` with

scripts/seed_system_configs.py

@@ -21,6 +21,11 @@
 from shared.clients.internal_api import InternalAPISyncClient
 
 CONFIG_PATH = Path(__file__).resolve().parent / "system_configs.yaml"
+_WORK_ADMISSION_KEYS = {
+    "work_admission.emergency_stop",
+    "work_admission.max_projects_per_user",
+    "work_admission.max_concurrent_paid_runs",
+}
 
 
 def load_configs(path: Path) -> list[dict]:
@@ -92,8 +97,15 @@ def seed_system_configs(api_base_url: str, configs_path: Path) -> bool:
                     "description": config["description"],
                     "updated_by": "seed",
                 }
-                resp = client.request_raw("POST", "system-configs/", json=payload)
-                if resp.status_code != httpx.codes.CREATED:
+                if key in _WORK_ADMISSION_KEYS:
+                    resp = client.request_raw(
+                        "PUT", f"work-admission/controls/{key}", json={"value": value}
+                    )
+                    expected_status = httpx.codes.OK
+                else:
+                    resp = client.request_raw("POST", "system-configs/", json=payload)
+                    expected_status = httpx.codes.CREATED
+                if resp.status_code != expected_status:
                     print(f"  Failed to write '{key}': {resp.status_code} - {resp.text}")
                     success = False
                     continue

scripts/system_configs.service_test.yaml

@@ -0,0 +1,20 @@
+# Overrides for the shared API service-test database.
+#
+# Service tests intentionally retain records across the suite to exercise APIs
+# against an evolving dataset. These ceilings remain enabled in the test
+# contour, but use explicit high values so unrelated scenarios do not exhaust
+# an operational production limit. Admission tests set their own low value.
+- key: work_admission.max_projects_per_user
+  value: 10000
+  category: work_admission
+  description: "Service-test maximum non-archived projects for one non-admin user"
+
+- key: work_admission.max_concurrent_paid_runs
+  value: 10000
+  category: work_admission
+  description: "Service-test maximum queued or running engineering and QA coding-agent runs"
+
+- key: work_admission.emergency_stop
+  value: false
+  category: work_admission
+  description: "Service-test operator switch that blocks new admitted work"

scripts/system_configs.yaml

@@ -6,6 +6,21 @@
 # the database alone and are never touched.
 
 # --- scheduler ---
+- key: work_admission.max_projects_per_user
+  value: 3
+  category: work_admission
+  description: "Maximum non-archived projects for one non-admin user"
+
+- key: work_admission.max_concurrent_paid_runs
+  value: 5
+  category: work_admission
+  description: "Maximum queued or running engineering and QA coding-agent runs"
+
+- key: work_admission.emergency_stop
+  value: false
+  category: work_admission
+  description: "Operator switch that blocks new projects, engineering and QA work"
+
 - key: scheduler.dispatch_interval_seconds
   value: 30
   category: scheduler

services/api/Dockerfile

@@ -42,6 +42,7 @@ COPY services/api/src ./src
 COPY services/api/migrations ./migrations
 COPY services/api/alembic.ini ./alembic.ini
 COPY services/api/entrypoint.sh /entrypoint.sh
+COPY scripts/seed_system_configs.py scripts/system_configs.yaml scripts/system_configs.service_test.yaml /app/scripts/
 RUN chmod +x /entrypoint.sh
 
 # Expose port

services/api/entrypoint.sh

@@ -4,4 +4,22 @@ set -e
 cd /app
 alembic upgrade head
 
+if [ "$ENVIRONMENT" = "test" ] && [ "$SEED_SYSTEM_CONFIGS_ON_START" = "true" ]; then
+    uvicorn src.main:app --host 0.0.0.0 --port 8000 &
+    api_pid=$!
+    until curl --fail --silent http://localhost:8000/health >/dev/null; do
+        sleep 1
+    done
+    python /app/scripts/seed_system_configs.py \
+        --api-base-url http://localhost:8000 \
+        --configs-path /app/scripts/system_configs.yaml
+    if [ -n "$SYSTEM_CONFIGS_TEST_OVERLAY" ]; then
+        python /app/scripts/seed_system_configs.py \
+            --api-base-url http://localhost:8000 \
+            --configs-path "$SYSTEM_CONFIGS_TEST_OVERLAY"
+    fi
+    wait "$api_pid"
+    exit $?
+fi
+
 exec uvicorn src.main:app --host 0.0.0.0 --port 8000

services/api/migrations/versions/a3d4e5f6a7b8_add_paid_admission_command_audit.py

@@ -0,0 +1,25 @@
+"""Persist paid-work command identity and the owner-facing refusal text.
+
+Revision ID: a3d4e5f6a7b8
+Revises: f7e8d9c0b1a2
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "a3d4e5f6a7b8"
+down_revision: str | None = "f7e8d9c0b1a2"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column("work_admission_audits", sa.Column("command_payload", sa.JSON(), nullable=True))
+    op.add_column("work_admission_audits", sa.Column("message", sa.Text(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("work_admission_audits", "message")
+    op.drop_column("work_admission_audits", "command_payload")

services/api/migrations/versions/f7e8d9c0b1a2_add_work_admission_audit.py

@@ -0,0 +1,44 @@
+"""Add count-based work admission audit records.
+
+Revision ID: f7e8d9c0b1a2
+Revises: f6e7d8c9b0a1
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "f7e8d9c0b1a2"
+down_revision: str | None = "f6e7d8c9b0a1"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "work_admission_audits",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("subject", sa.String(length=32), nullable=False),
+        sa.Column("outcome", sa.String(length=16), nullable=False),
+        sa.Column("reason", sa.String(length=32), nullable=True),
+        sa.Column("user_id", sa.Integer(), nullable=True),
+        sa.Column("reference_id", sa.String(length=255), nullable=True),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index("ix_work_admission_audits_subject", "work_admission_audits", ["subject"])
+    op.create_index("ix_work_admission_audits_outcome", "work_admission_audits", ["outcome"])
+    op.create_index("ix_work_admission_audits_user_id", "work_admission_audits", ["user_id"])
+    op.create_index(
+        "ix_work_admission_audits_reference_id", "work_admission_audits", ["reference_id"]
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("work_admission_audits")

services/api/src/main.py

@@ -147,6 +147,7 @@ async def root():
 app.include_router(routers.runs.router, prefix="/api")
 app.include_router(routers.engineering_budget_policies.router, prefix="/api")
 app.include_router(routers.engineering_budget_policies.self_router, prefix="/api")
+app.include_router(routers.work_admission.router, prefix="/api")
 app.include_router(routers.tasks.router, prefix="/api")
 app.include_router(routers.brainstorms.router, prefix="/api")
 app.include_router(routers.repositories.router, prefix="/api")

services/api/src/routers/__init__.py

@@ -25,6 +25,7 @@
     tasks,
     temporary_access,
     users,
+    work_admission,
 )
 
 __all__ = [
@@ -52,4 +53,5 @@
     "tasks",
     "temporary_access",
     "users",
+    "work_admission",
 ]