refactor(core): lift create verb body across both kinds (Stage 2 verb lift)

New ``make_create_handler(cfg, *, audit_emit=None)`` factory in
``turnstone/core/session_routes.py`` consumes five new ``SessionEndpointConfig``
fields (``create_supports_attachments``, ``create_supports_user_id_override``,
``create_validate_request``, ``create_build_kwargs``, ``create_post_install``)
and replaces both ``create_workstream`` and ``coordinator_create`` bodies.
Same factory + capability-flag pattern as the merged cancel / open / events
lifts. ``_validate_and_save_uploaded_files`` lifted to
``turnstone.core.attachments`` so both processes call one kind-agnostic
implementation.

Coord parity gains (§ Post-P3 reckoning item #1 + carry-forward):
- Create-time attachments: multipart parsing, validate+save+rollback,
  ``attachment_ids`` on the response. Coord adapter ``send`` doesn't yet
  reserve attachments at create time, so the rows save as pending and the
  next ``/send`` picks them up via the standard send-with-attachments path.
- Disabled-skill rejection (matches interactive's pre-lift gate).
- Always-include response shape ``{ws_id, name, resumed, message_count,
  attachment_ids}`` populated with default ``False``/``0``/``[]`` on the
  fields coord doesn't fill.
- 200 status (was 201).
- Audit-emit failures swallow + warning log instead of 500.

Both kinds converge on the manager-at-capacity 429, factory-misconfig 503,
and correlation_id'd 500 for unexpected ``mgr.create`` failure (interactive
lifted up to coord's safer error envelope).

Three /review fixes folded in:
- ``notify_targets`` malformed input gates at the validator (400) instead
  of bubbling out of post_install as a 500 — pre-fix the workstream had
  already been created + audited + broadcast by the time the validation
  raised.
- Skill-lookup storage failures now share the correlation_id'd 500 path
  with ``mgr.create`` (was masquerading as 400 "Skill not found").
- Whitespace-only ``skill`` field treated as empty (matches pre-lift coord).

CHANGELOG entry under [Unreleased] documents every observable behaviour
change. OpenAPI spec regenerated. Three new coord regression tests
(create-time-attachments save pending rows, always-include parity fields,
disabled-skill rejection) plus one interactive regression test
(notify_targets 400). 4500 tests passing.

CHANGELOG.md

@@ -400,6 +400,151 @@ Three release tracks are maintained:
     semantics are preserved by wrapping the iteration in the same
     try/except.
 
+- **`create` verb body lifted across both kinds** ([Stage 2 Verb
+  Lift — `create`]). The interactive
+  ``POST /v1/api/workstreams/new`` and coord
+  ``POST /v1/api/workstreams/new`` handlers now share one body via
+  ``make_create_handler(cfg, *, audit_emit=None)``. Per-kind
+  divergence captured by five new ``SessionEndpointConfig`` fields:
+
+  - ``create_supports_attachments: bool`` — multipart body parsing
+    + attachment validation+save+rollback. Both kinds wire ``True``.
+  - ``create_supports_user_id_override: bool`` — trusted-source
+    body ``user_id`` override (interactive ``True`` for console-
+    proxied creates; coord ``False``).
+  - ``create_validate_request: CreateRequestValidator | None`` —
+    per-kind pre-create gates (interactive: ws_id format, kind,
+    parent ownership, attachments+resume_ws combo; coord: 401-on-
+    empty-uid).
+  - ``create_build_kwargs: CreateKwargsBuilder | None`` — per-kind
+    kwargs dict for ``mgr.create``.
+  - ``create_post_install: CreatePostInstall | None`` — per-kind
+    tail end (interactive: WebUI auto_approve + watch_runner +
+    ``ws_created`` global broadcast + atomic resume + skill session
+    config + notify_targets + routing override + initial-message
+    worker thread; coord: ``coord_adapter.send`` for the optional
+    initial_message).
+
+  The pure helper ``_validate_and_save_uploaded_files`` lifted from
+  ``turnstone.server`` to ``turnstone.core.attachments`` as
+  ``validate_and_save_uploaded_files`` so both processes can call
+  the same kind-agnostic implementation.
+
+  **§ Post-P3 reckoning item #1 done — coord gains create-time
+  attachments.** Pre-lift ``coordinator_create`` accepted JSON only
+  and ignored uploads; the lifted body parses ``multipart/form-data``
+  on coord and saves attachments through the kind-agnostic storage
+  layer. Coord's adapter ``send`` does NOT yet take attachments,
+  so when a create request carries both ``initial_message`` and
+  uploads, the rows save as pending and the next ``/send`` picks
+  them up via the standard send-with-attachments path. Initial-
+  message + create-time attachments coordination on coord is a
+  follow-up — out of scope for the verb-shape lift.
+
+  Note on broadcast timing: coord's ``mgr.create`` fires
+  ``emit_created`` (cluster collector fan-out) BEFORE the lifted
+  body runs attachment validation. If validation fails on coord and
+  the rollback (``mgr.close`` → ``emit_closed``) fires, the cluster
+  events stream sees a phantom create→close pair. Cluster consumers
+  handle this gracefully (same shape as any quick-create-close);
+  decoupling ``emit_created`` from ``mgr.create`` would be a bigger
+  refactor that doesn't belong in the verb lift. Interactive's
+  broadcast (``gq.put_nowait("ws_created")``) is held until after
+  attachment validation by the post-install callback, so interactive
+  never sees the phantom pair.
+
+  Five observable behaviour changes on the create response:
+
+  - **Both kinds converge on 200 OK.** Pre-lift interactive
+    returned 200 (default JSONResponse status); pre-lift coord
+    returned 201. Picked 200 over 201 for response-shape parity
+    with every other shared verb at the cost of REST-strict
+    correctness — a one-time release note rather than ongoing
+    client churn (the rest of the v1 SDK already uses
+    ``response.ok`` per ``feedback_test_frontend_locally.md``).
+    SDK consumers that branched on ``status == 201`` for coord
+    must switch to ``response.ok``.
+  - **Always-include response shape.** Pre-lift interactive
+    returned ``{ws_id, name, resumed, message_count, attachment_ids}``
+    (5 fields); pre-lift coord returned ``{ws_id, name}`` (2). The
+    lifted body always returns the full shape, with ``resumed=False``
+    / ``message_count=0`` / ``attachment_ids=[]`` on kinds whose
+    post-install doesn't populate them. Coord callers will see the
+    parity fields appear with default values.
+  - **Both kinds converge on the manager-at-capacity 429
+    semantic.** Pre-lift interactive translated ``mgr.create``'s
+    ``RuntimeError`` to 400; coord already translated to 429. The
+    documented contract on ``SessionManager.create`` is "raises
+    RuntimeError when the manager is at capacity" — 429 (rate-
+    limit / try-later) is the correct shape.
+  - **Both kinds converge on the factory-misconfig 503 semantic.**
+    Pre-lift interactive let ``ValueError`` propagate as 500 with
+    a stack trace; coord already translated to 503 with the
+    factory's remediation text. Operators get the actionable
+    message instead of the trace.
+  - **Both kinds get a correlation_id'd 500 on unexpected
+    ``mgr.create`` failure.** Pre-lift interactive let unexpected
+    exceptions propagate as 500 with a stack trace (potential
+    information leak via frame names / file paths); coord already
+    returned a correlation_id'd 500 with the message redacted. The
+    lifted body adopts coord's safer pattern on both kinds.
+
+  Two coord-specific parity gains:
+
+  - **Coord rejects disabled skills.** Pre-lift
+    ``coordinator_create`` silently allowed disabled skills to
+    flow through to ``mgr.create`` — the row would create with a
+    skill the operator had marked inert, surprising both the
+    operator and the next user. The lifted body returns 400
+    "Skill not found or disabled" matching interactive's
+    behaviour.
+  - **Coord audit-emit failures no longer 500.** Pre-lift
+    ``coordinator_create`` already swallowed; pre-lift interactive
+    let the failure propagate as 500. The lifted body wraps
+    ``audit_emit`` in try/except + ``warning`` log, returning the
+    successful 200 to the caller. Mirrors the close / cancel /
+    open / events lift contracts.
+
+  No legacy adapter is needed for create — both kinds already
+  mounted ``POST {prefix}/new`` pre-lift; the lifted handler slots
+  in at the same path on each kind.
+
+  Three /review fixes folded into the same commit:
+
+  - **Pre-lift's 400 on malformed ``notify_targets`` preserved.** The
+    initial draft surfaced ``notify_targets`` validation errors from
+    inside the interactive ``post_install`` callback, which the
+    factory had no return-the-400 channel for — the only signal was
+    to ``raise``, which the factory's generic exception handler
+    turned into a redacted 500. Worse, by the time ``post_install``
+    ran the workstream was fully built (audit row written,
+    ``ws_created`` broadcast emitted), so a malformed-input request
+    surfaced as "create failed" with the workstream actually live.
+    Fixed by moving the ``notify_targets`` validation into
+    :func:`_interactive_create_validate_request` (the pre-create
+    gate), which returns the 400 before ``mgr.create`` runs and
+    keeps storage clean. New regression test:
+    ``test_create_lift_400s_on_malformed_notify_targets``.
+  - **Skill-lookup storage failure now correlation_id'd.** The
+    initial draft swallowed ``get_skill_by_name`` exceptions into
+    ``skill_data = None`` and returned a 400 "Skill not found or
+    disabled" — masking storage outages as user-input misses and
+    making operator triage of skill-related reports impossible. The
+    lifted body now lets the storage exception propagate to the
+    same correlation_id'd 500 path that ``mgr.create`` failures
+    use; the skill-lookup + version count + ``mgr.create`` all live
+    inside one ``try / except`` so storage outages anywhere in the
+    create-prelude get the redacted-message-with-correlation-id
+    treatment instead of a stack-traced 500 leak.
+  - **Whitespace-only ``skill`` field treated as empty.** The
+    initial draft took ``body.get("skill") or ""`` literally — a
+    payload with ``"skill": "  "`` would have hit
+    ``get_skill_by_name(" ")`` and 400'd as "Skill not found".
+    Pre-lift coord stripped via ``(body.get("skill") or "").strip()
+    or None``; the lifted body now strips for both kinds (interactive
+    never received whitespace-only skills from the web UI but the
+    convergence is the safer default).
+
 ### Security
 
 - **Coord attachment endpoints are now kind-strict**

sdk/typescript/openapi-console.json

@@ -4504,7 +4504,7 @@
         "tags": [
           "Coordinator"
         ],
-        "description": "Allocates a console-hosted ``kind=\"coordinator\"`` ChatSession.  201 on create; 429 when the ``coordinator.max_active`` cap is reached and no idle coordinator can be evicted.",
+        "description": "Allocates a console-hosted ``kind=\"coordinator\"`` ChatSession.  200 on create; 429 when the ``coordinator.max_active`` cap is reached and no idle coordinator can be evicted.  Pre-1.5.0 this returned 201; the lifted ``create`` factory (Stage 2 verb lift) converges on 200 across both kinds.",
         "requestBody": {
           "required": true,
           "content": {
@@ -4516,7 +4516,7 @@
           }
         },
         "responses": {
-          "201": {
+          "200": {
             "description": "Success",
             "content": {
               "application/json": {
@@ -7458,7 +7458,7 @@
         "type": "object"
       },
       "CoordinatorCreateResponse": {
-        "description": "Response body for POST /v1/api/workstreams/new (201).",
+        "description": "Response body for POST /v1/api/workstreams/new (200).",
         "properties": {
           "ws_id": {
             "title": "Ws Id",
@@ -7467,6 +7467,23 @@
           "name": {
             "title": "Name",
             "type": "string"
+          },
+          "resumed": {
+            "default": false,
+            "title": "Resumed",
+            "type": "boolean"
+          },
+          "message_count": {
+            "default": 0,
+            "title": "Message Count",
+            "type": "integer"
+          },
+          "attachment_ids": {
+            "items": {
+              "type": "string"
+            },
+            "title": "Attachment Ids",
+            "type": "array"
           }
         },
         "required": [

tests/test_coordinator_end_to_end.py

@@ -32,9 +32,12 @@ from turnstone.console.coordinator_client import CoordinatorClient
 from turnstone.console.coordinator_ui import ConsoleCoordinatorUI
 from turnstone.console.server import (
     _audit_close_coordinator,
+    _audit_coordinator_create,
+    _coord_create_build_kwargs,
+    _coord_create_post_install,
+    _coord_create_validate_request,
     _require_admin_coordinator,
     _require_coord_mgr,
-    coordinator_create,
     coordinator_detail,
     coordinator_list,
 )
@@ -43,6 +46,7 @@ from turnstone.core.session_manager import SessionManager
 from turnstone.core.session_routes import (
     SessionEndpointConfig,
     make_close_handler,
+    make_create_handler,
 )
 from turnstone.core.storage._sqlite import SQLiteBackend
 
@@ -53,6 +57,11 @@ _coord_endpoint_config = SessionEndpointConfig(
     tenant_check=None,
     not_found_label="coordinator not found",
     audit_action_prefix="coordinator",
+    create_supports_attachments=True,
+    create_supports_user_id_override=False,
+    create_validate_request=_coord_create_validate_request,
+    create_build_kwargs=_coord_create_build_kwargs,
+    create_post_install=_coord_create_post_install,
 )
 
 
@@ -136,7 +145,7 @@ def _make_client(
         routes=[
             Route(
                 "/v1/api/workstreams/new",
-                coordinator_create,
+                make_create_handler(_coord_endpoint_config, audit_emit=_audit_coordinator_create),
                 methods=["POST"],
             ),
             Route("/v1/api/workstreams", coordinator_list, methods=["GET"]),
@@ -186,7 +195,7 @@ def test_create_list_detail_lifecycle(tmp_path):
         json={"name": "e2e-coord"},
         headers=_COORD_HEADERS,
     )
-    assert resp.status_code == 201, resp.text
+    assert resp.status_code == 200, resp.text
     body = resp.json()
     ws_id = body["ws_id"]
     assert ws_id

tests/test_coordinator_endpoints.py

@@ -33,11 +33,14 @@ from turnstone.console.coordinator_ui import ConsoleCoordinatorUI
 from turnstone.console.server import (
     _audit_cancel_coordinator,
     _audit_close_coordinator,
+    _audit_coordinator_create,
+    _coord_create_build_kwargs,
+    _coord_create_post_install,
+    _coord_create_validate_request,
     _require_admin_coordinator,
     _require_coord_mgr,
     cluster_ws_detail,
     coordinator_children,
-    coordinator_create,
     coordinator_detail,
     coordinator_history,
     coordinator_list,
@@ -61,6 +64,7 @@ from turnstone.core.session_routes import (
     make_attachment_handlers,
     make_cancel_handler,
     make_close_handler,
+    make_create_handler,
     make_open_handler,
     make_send_handler,
 )
@@ -106,6 +110,11 @@ _coord_endpoint_config = SessionEndpointConfig(
     ),
     spawn_metrics=None,
     emit_message_queued=True,
+    create_supports_attachments=True,
+    create_supports_user_id_override=False,
+    create_validate_request=_coord_create_validate_request,
+    create_build_kwargs=_coord_create_build_kwargs,
+    create_post_install=_coord_create_post_install,
 )
 
 
@@ -123,11 +132,14 @@ def _make_client(
 ) -> TestClient:
     """Build a TestClient exposing just the coordinator routes."""
     coord_attachments = make_attachment_handlers(_coord_endpoint_config)
+    coord_create_handler = make_create_handler(
+        _coord_endpoint_config, audit_emit=_audit_coordinator_create
+    )
     app = Starlette(
         routes=[
             Route(
                 "/v1/api/workstreams/new",
-                coordinator_create,
+                coord_create_handler,
                 methods=["POST"],
             ),
             Route("/v1/api/workstreams", coordinator_list, methods=["GET"]),
@@ -334,19 +346,125 @@ def test_create_returns_ws_id_and_records_audit(storage):
         json={"name": "my-coord"},
         headers=_COORD_HEADERS,
     )
-    assert resp.status_code == 201
+    assert resp.status_code == 200
     body = resp.json()
     assert body["ws_id"]
     assert "my-coord" in body["name"]
+    # Always-include parity fields land on coord post-`create` lift —
+    # SDK consumers don't have to branch on kind to read them.
+    assert body["resumed"] is False
+    assert body["message_count"] == 0
+    assert body["attachment_ids"] == []
     # Audit row recorded on storage.
-    from turnstone.core.audit import record_audit  # noqa: F401 (verify import works)
-
-    # Query audit_events via storage.
     events = storage.list_audit_events(user_id="user-1", limit=10)
     actions = [e["action"] for e in events]
     assert "coordinator.create" in actions
 
 
+def test_create_with_multipart_attachments_saves_pending_rows(storage):
+    """§ Post-P3 reckoning item #1 regression — coord gains create-time
+    attachments. Multipart create with a magic-byte-valid PNG saves
+    a pending attachment row scoped to the new coord ws_id."""
+    from turnstone.core.memory import list_pending_attachments
+
+    # Magic-byte-valid 1x1 PNG (matches test_server_attachments_endpoints.py).
+    png_1x1 = (
+        b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
+        b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\rIDATx\x9cc\xfc\xcf"
+        b"\xc0\xc0\xc0\x00\x00\x00\x05\x00\x01\xa5\xf6E@\x00\x00\x00\x00IEND"
+        b"\xaeB`\x82"
+    )
+    mgr = _build_mgr(storage)
+    client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+
+    # Inject the test storage backend as the global singleton so
+    # ``save_attachment`` / ``list_pending_attachments`` (which both
+    # go through ``turnstone.core.memory`` → ``get_storage()``)
+    # resolve onto our SQLiteBackend instead of the real one.
+    import turnstone.core.storage._registry as _reg
+
+    _old_storage = _reg._storage
+    _reg._storage = storage
+    try:
+        resp = client.post(
+            "/v1/api/workstreams/new",
+            data={"meta": '{"name": "with-image"}'},
+            files={"file": ("img.png", png_1x1, "image/png")},
+            headers=_COORD_HEADERS,
+        )
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        ws_id = body["ws_id"]
+        assert ws_id
+        assert len(body["attachment_ids"]) == 1
+        # Attachment row is pending (saved but not consumed) — coord's
+        # adapter ``send`` doesn't reserve attachments at create time
+        # yet, so the next ``/send`` picks it up via the standard
+        # send-with-attachments path.
+        pending = list_pending_attachments(ws_id, "user-1")
+        assert len(pending) == 1
+        assert pending[0]["kind"] == "image"
+    finally:
+        _reg._storage = _old_storage
+
+
+def test_create_rejects_disabled_skill(storage):
+    """Coord parity gain — disabled skills now rejected at the lift,
+    matching interactive's pre-lift behaviour. Pre-lift coord silently
+    let disabled skills through."""
+    import turnstone.core.storage._registry as _reg
+
+    _old_storage = _reg._storage
+    _reg._storage = storage
+    try:
+        # Save a disabled skill row so ``get_skill_by_name`` resolves
+        # but the lifted body's enabled check rejects it. Mirrors the
+        # ``_create_template`` helper in ``tests/test_skills.py``;
+        # inlined here so this regression test stays self-contained.
+        storage.create_prompt_template(
+            template_id="sk-disabled",
+            name="dormant-skill",
+            category="general",
+            content="dormant",
+            variables="[]",
+            is_default=False,
+            org_id="",
+            created_by="test",
+            origin="manual",
+            mcp_server="",
+            readonly=False,
+            description="",
+            tags="[]",
+            source_url="",
+            version="1.0.0",
+            author="",
+            activation="named",
+            token_estimate=0,
+            model="",
+            auto_approve=False,
+            temperature=None,
+            reasoning_effort="",
+            max_tokens=None,
+            token_budget=0,
+            agent_max_turns=None,
+            notify_on_complete="{}",
+            enabled=False,  # the gate under test
+            allowed_tools="[]",
+            priority=0,
+        )
+        mgr = _build_mgr(storage)
+        client = _make_client(storage, coord_mgr=mgr, registry=_fake_registry())
+        resp = client.post(
+            "/v1/api/workstreams/new",
+            json={"name": "x", "skill": "dormant-skill"},
+            headers=_COORD_HEADERS,
+        )
+        assert resp.status_code == 400
+        assert "dormant-skill" in resp.json()["error"]
+    finally:
+        _reg._storage = _old_storage
+
+
 def test_list_returns_cluster_wide(storage):
     # Trusted-team visibility: any caller with admin.coordinator sees
     # every active coordinator regardless of owner.  ``user_id`` stays

tests/test_openapi.py

@@ -139,16 +139,20 @@ class TestConsoleSpec:
         }
         assert expected.issubset(paths), f"Missing: {expected - paths}"
 
-    def test_coordinator_create_has_request_body_and_201(self):
-        """Coordinator create returns 201 (not 200) and accepts a body."""
+    def test_coordinator_create_has_request_body_and_200(self):
+        """Coordinator create returns 200 and accepts a body.
+
+        Pre-1.5.0 this returned 201 (REST-strict for create); the lifted
+        ``make_create_handler`` factory converges on 200 across both
+        kinds for response-shape parity with every other shared verb.
+        """
         from turnstone.api.console_spec import build_console_spec
 
         spec = build_console_spec()
         op = spec["paths"]["/v1/api/workstreams/new"]["post"]
         assert "requestBody" in op
         assert "application/json" in op["requestBody"]["content"]
-        # Pin the 201 success code.
-        assert "201" in op["responses"]
+        assert "200" in op["responses"]
 
     def test_coordinator_history_has_limit_query_param(self):
         from turnstone.api.console_spec import build_console_spec

tests/test_server_attachments_on_create.py

@@ -50,9 +50,11 @@ def _auth(user: str) -> dict[str, str]:
 
 class TestValidateAndSaveUploadedFiles:
     def test_saves_image_and_text(self, tmp_path):
+        from turnstone.core.attachments import (
+            validate_and_save_uploaded_files as _validate_and_save_uploaded_files,
+        )
         from turnstone.core.memory import list_pending_attachments
         from turnstone.core.storage import init_storage, reset_storage
-        from turnstone.server import _validate_and_save_uploaded_files
 
         reset_storage()
         init_storage("sqlite", path=str(tmp_path / "t.db"), run_migrations=False)
@@ -73,8 +75,10 @@ class TestValidateAndSaveUploadedFiles:
 
     def test_rejects_oversized_image(self, tmp_path):
         from turnstone.core.attachments import IMAGE_SIZE_CAP
+        from turnstone.core.attachments import (
+            validate_and_save_uploaded_files as _validate_and_save_uploaded_files,
+        )
         from turnstone.core.storage import init_storage, reset_storage
-        from turnstone.server import _validate_and_save_uploaded_files
 
         reset_storage()
         init_storage("sqlite", path=str(tmp_path / "t.db"), run_migrations=False)
@@ -90,8 +94,10 @@ class TestValidateAndSaveUploadedFiles:
             reset_storage()
 
     def test_rejects_unsupported_text(self, tmp_path):
+        from turnstone.core.attachments import (
+            validate_and_save_uploaded_files as _validate_and_save_uploaded_files,
+        )
         from turnstone.core.storage import init_storage, reset_storage
-        from turnstone.server import _validate_and_save_uploaded_files
 
         reset_storage()
         init_storage("sqlite", path=str(tmp_path / "t.db"), run_migrations=False)
@@ -107,9 +113,11 @@ class TestValidateAndSaveUploadedFiles:
 
     def test_pending_cap_returns_409(self, tmp_path):
         from turnstone.core.attachments import MAX_PENDING_ATTACHMENTS_PER_USER_WS
+        from turnstone.core.attachments import (
+            validate_and_save_uploaded_files as _validate_and_save_uploaded_files,
+        )
         from turnstone.core.memory import save_attachment
         from turnstone.core.storage import init_storage, reset_storage
-        from turnstone.server import _validate_and_save_uploaded_files
 
         reset_storage()
         init_storage("sqlite", path=str(tmp_path / "t.db"), run_migrations=False)

tests/test_skills.py

@@ -1731,7 +1731,18 @@ class TestSkillConfigAppliedToWorkstream:
         import turnstone.core.storage._registry as _reg
         from turnstone.core.adapters.interactive_adapter import InteractiveAdapter
         from turnstone.core.session_manager import SessionManager
-        from turnstone.server import WebUI, create_workstream
+        from turnstone.core.session_routes import (
+            SessionEndpointConfig,
+            make_create_handler,
+        )
+        from turnstone.server import (
+            WebUI,
+            _interactive_create_build_kwargs,
+            _interactive_create_post_install,
+            _interactive_create_validate_request,
+            _interactive_manager_lookup,
+            _interactive_tenant_check,
+        )
 
         storage = SQLiteBackend(str(tmp_path / "ws_test.db"))
 
@@ -1769,13 +1780,29 @@ class TestSkillConfigAppliedToWorkstream:
         )
         mgr = SessionManager(adapter, storage=storage, max_active=10, event_emitter=adapter)
 
+        # Build the same lifted create handler the production app
+        # mounts so this fixture exercises the make_create_handler
+        # factory rather than a parallel pre-lift body.
+        _test_cfg = SessionEndpointConfig(
+            permission_gate=None,
+            manager_lookup=_interactive_manager_lookup,
+            tenant_check=_interactive_tenant_check,
+            not_found_label="Workstream not found",
+            audit_action_prefix="workstream",
+            create_supports_attachments=True,
+            create_supports_user_id_override=True,
+            create_validate_request=_interactive_create_validate_request,
+            create_build_kwargs=_interactive_create_build_kwargs,
+            create_post_install=_interactive_create_post_install,
+        )
+        _test_create_handler = make_create_handler(_test_cfg)
         routes = [
             Mount(
                 "/v1",
                 routes=[
                     Route(
                         "/api/workstreams/new",
-                        create_workstream,
+                        _test_create_handler,
                         methods=["POST"],
                     ),
                 ],
@@ -1798,6 +1825,23 @@ class TestSkillConfigAppliedToWorkstream:
         # Restore original storage singleton.
         _reg._storage = old_storage
 
+    def test_create_lift_400s_on_malformed_notify_targets(self, _ws_app):
+        """Regression for the lifted create handler — malformed
+        ``notify_targets`` returns 400 from the validator (pre-create
+        gate), not 500 from a post_install raise."""
+        client, mgr, storage = _ws_app
+
+        resp = client.post(
+            "/v1/api/workstreams/new",
+            json={"name": "x", "notify_targets": "{not json"},
+        )
+        assert resp.status_code == 400, resp.text
+        body = resp.json()
+        assert "error" in body
+        # The workstream must NOT have been created — the validator
+        # gates BEFORE mgr.create, so storage stays clean.
+        assert len(list(storage.list_workstreams())) == 0
+
     def test_session_receives_temperature(self, _ws_app):
         """Skill temperature overrides the session default."""
         client, mgr, storage = _ws_app

turnstone/api/console_schemas.py

@@ -1010,10 +1010,18 @@ class CoordinatorCreateRequest(BaseModel):
 
 
 class CoordinatorCreateResponse(BaseModel):
-    """Response body for POST /v1/api/workstreams/new (201)."""
+    """Response body for POST /v1/api/workstreams/new (200)."""
 
     ws_id: str
     name: str
+    # Always-include parity fields from the Stage 2 ``create`` verb
+    # lift. Coord doesn't populate ``resumed`` or ``message_count``
+    # today (no resume-on-create surface yet), so they default to
+    # ``False`` / ``0``. ``attachment_ids`` carries the saved-but-
+    # pending attachment ids when the request was multipart.
+    resumed: bool = False
+    message_count: int = 0
+    attachment_ids: list[str] = Field(default_factory=list)
 
 
 class CoordinatorInfo(BaseModel):

turnstone/api/console_spec.py

@@ -1140,12 +1140,14 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [
         "Create a new coordinator workstream",
         description=(
             'Allocates a console-hosted ``kind="coordinator"`` ChatSession.  '
-            "201 on create; 429 when the ``coordinator.max_active`` cap is "
-            "reached and no idle coordinator can be evicted."
+            "200 on create; 429 when the ``coordinator.max_active`` cap is "
+            "reached and no idle coordinator can be evicted.  "
+            "Pre-1.5.0 this returned 201; the lifted ``create`` factory "
+            "(Stage 2 verb lift) converges on 200 across both kinds."
         ),
         request_model=CoordinatorCreateRequest,
         response_model=CoordinatorCreateResponse,
-        response_code=201,
+        response_code=200,
         error_codes=[400, 401, 403, 429, 500, 503],
         tags=["Coordinator"],
     ),

turnstone/console/server.py

@@ -63,6 +63,7 @@ from turnstone.core.session_routes import (
     make_attachment_handlers,
     make_cancel_handler,
     make_close_handler,
+    make_create_handler,
     make_events_handler,
     make_open_handler,
     make_send_handler,
@@ -2481,110 +2482,111 @@ def _coord_events_replay(
         yield pending_plan
 
 
-async def coordinator_create(request: Request) -> JSONResponse:
-    """POST /v1/api/workstreams/new — create a new coordinator session."""
-    from turnstone.core.audit import record_audit
-    from turnstone.core.web_helpers import read_json_or_400
+async def _coord_create_validate_request(
+    request: Request,
+    body: dict[str, Any],
+    uid: str,
+    uploaded_files: list[tuple[str, str, bytes]],
+) -> JSONResponse | None:
+    """Per-kind pre-create gate for coord.
 
-    err = _require_admin_coordinator(request)
-    if err is not None:
-        return err
-    coord_mgr, err503 = _require_coord_mgr(request)
-    if err503 is not None:
-        return err503
-    body = await read_json_or_400(request)
-    if isinstance(body, JSONResponse):
-        return body
-    user_id = _auth_user_id(request)
-    if not user_id:
+    Wired onto :attr:`SessionEndpointConfig.create_validate_request`
+    and called by :func:`make_create_handler` after body parsing
+    but before skill resolution / ``mgr.create``. The single gate is
+    a 401 when the auth result resolved to an empty user id —
+    coord's ``admin.coordinator`` scope check at the
+    ``permission_gate`` is the primary access boundary, but a token
+    that passes the scope check with ``sub=""`` would still land
+    here, and ``mgr.create`` requires a non-empty ``user_id``.
+    """
+    if not uid:
         return JSONResponse({"error": "authentication required"}, status_code=401)
-    name = (body.get("name") or "").strip()
-    skill = (body.get("skill") or "").strip() or None
-    initial_message = (body.get("initial_message") or "").strip()
-    # Pre-resolve skill → (template_id, applied_version) so the
-    # persisted row records what was applied. SessionManager.create
-    # is kind-agnostic and no longer does this lookup itself.
-    skill_id_resolved = ""
-    skill_version_resolved = 0
-    if skill:
-        from turnstone.core.memory import get_skill_by_name
-        from turnstone.core.storage._registry import get_storage as _get_storage
+    return None
 
-        _st = _get_storage()
-        skill_data = await asyncio.to_thread(get_skill_by_name, skill)
-        if skill_data and skill_data.get("template_id") and _st is not None:
-            skill_id_resolved = str(skill_data["template_id"])
-            try:
-                skill_version_resolved = (
-                    await asyncio.to_thread(_st.count_skill_versions, skill_id_resolved) + 1
-                )
-            except Exception:
-                log.debug("coord_create.skill_version_failed skill=%s", skill, exc_info=True)
-                skill_version_resolved = 1
-    try:
-        # Offload to a worker thread — SessionManager.create runs
-        # blocking storage calls (register_workstream) and the
-        # session-factory invocation. Running inline on the async
-        # event loop stalled the SSE manager + every other async
-        # handler for the duration of the create (#perf-4).
-        ws = await asyncio.to_thread(
-            coord_mgr.create,
-            user_id=user_id,
-            name=name,
-            skill=skill,
-            skill_id=skill_id_resolved,
-            skill_version=skill_version_resolved,
-        )
-    except RuntimeError as exc:
-        return JSONResponse({"error": str(exc)}, status_code=429)
-    except ValueError as exc:
-        # Session factory raises ValueError on misconfigured alias —
-        # surface as 503 with the factory's remediation text.
-        return JSONResponse({"error": str(exc)}, status_code=503)
-    except Exception:
-        # Don't echo the exception text to the caller — it can leak
-        # internals (stack frame names, file paths, etc.).  Log with a
-        # short correlation id and return that to the client so support
-        # can match a user report to the log line.
-        correlation_id = secrets.token_hex(4)
-        log.warning(
-            "coordinator_create.failed correlation_id=%s",
-            correlation_id,
-            exc_info=True,
-        )
-        return JSONResponse(
-            {
-                "error": (
-                    "failed to create coordinator (internal error). "
-                    f"correlation_id={correlation_id}"
-                )
-            },
-            status_code=500,
-        )
-    # Initial message spawns the first worker after the session has
-    # been installed on the adapter. Previously this rode inside
-    # ``coord_mgr.create`` via the old ``initial_message`` kwarg; the
-    # unified SessionManager.create is kind-agnostic so the worker spawn
-    # moved to the caller.
+
+def _coord_create_build_kwargs(
+    request: Request,
+    body: dict[str, Any],
+    uid: str,
+    skill_data: dict[str, Any] | None,
+    skill_id: str,
+    applied_skill_version: int,
+) -> dict[str, Any]:
+    """Build kwargs for ``coord_mgr.create`` from a parsed coord create body.
+
+    Coord's create takes a smaller set than interactive's (no
+    ``model`` / ``judge_model`` / ``client_type`` / ``parent_ws_id`` /
+    ``ws_id``) — those concepts either don't apply to coordinators
+    (no parent on coord; coord ws_id is always server-generated)
+    or live on a separate ConfigStore knob (the dashboard-managed
+    plan/task model + reasoning_effort settings).
+    """
+    body_skill = (body.get("skill") or "").strip() or None
+    name = (body.get("name") or "").strip()
+    return {
+        "user_id": uid,
+        "name": name,
+        "skill": body_skill,
+        "skill_id": skill_id,
+        "skill_version": applied_skill_version,
+    }
+
+
+async def _coord_create_post_install(
+    request: Request,
+    ws: Workstream,
+    body: dict[str, Any],
+    uid: str,
+    skill_data: dict[str, Any] | None,
+    applied_skill_version: int,
+    attachment_ids: list[str],
+) -> dict[str, Any]:
+    """Tail end of coord create: dispatch the initial message.
+
+    Wired onto :attr:`SessionEndpointConfig.create_post_install`. The
+    factory has already saved any uploaded attachments through the
+    kind-agnostic storage layer; coord's adapter ``send`` does not
+    yet take attachments, so the rows save as pending and the next
+    ``/send`` picks them up via the standard send-with-attachments
+    path. Initial-message + create-time attachments coordination is
+    a follow-up.
+
+    Returns ``{}`` — coord's response carries only the always-include
+    parity fields populated by the factory.
+    """
+    initial_message = (body.get("initial_message") or "").strip()
     if initial_message:
         coord_adapter = getattr(request.app.state, "coord_adapter", None)
         if coord_adapter is not None:
             coord_adapter.send(ws.id, initial_message)
+    return {}
+
+
+def _audit_coordinator_create(
+    request: Request,
+    ws: Workstream,
+    body: dict[str, Any],
+    uid: str,
+) -> None:
+    """Audit emitter for the coord ``coordinator.create`` event.
+
+    Wired onto :func:`make_create_handler` as ``audit_emit``. Failures
+    are caught + logged at ``warning`` by the factory.
+    """
+    from turnstone.core.audit import record_audit
+
     storage = getattr(request.app.state, "auth_storage", None)
-    if storage is not None:
-        try:
-            record_audit(
-                storage,
-                user_id,
-                "coordinator.create",
-                "workstream",
-                ws.id,
-                {"coord_ws_id": ws.id, "src": "coordinator", "name": ws.name},
-                request.client.host if request.client else "",
-            )
-        except Exception:
-            log.debug("coordinator_create.audit_failed", exc_info=True)
-    return JSONResponse({"ws_id": ws.id, "name": ws.name}, status_code=201)
+    if storage is None:
+        return
+    record_audit(
+        storage,
+        uid,
+        "coordinator.create",
+        "workstream",
+        ws.id,
+        {"coord_ws_id": ws.id, "src": "coordinator", "name": ws.name},
+        request.client.host if request.client else "",
+    )
 
 
 async def coordinator_history(request: Request) -> JSONResponse:
@@ -10105,6 +10107,11 @@ def create_app(
         spawn_metrics=None,
         emit_message_queued=True,
         events_replay=_coord_events_replay,
+        create_supports_attachments=True,
+        create_supports_user_id_override=False,
+        create_validate_request=_coord_create_validate_request,
+        create_build_kwargs=_coord_create_build_kwargs,
+        create_post_install=_coord_create_post_install,
     )
     coord_workstream_routes: list[Any] = []
     register_session_routes(
@@ -10113,7 +10120,10 @@ def create_app(
         handlers=SharedSessionVerbHandlers(
             list_workstreams=coordinator_list,
             list_saved=coordinator_saved,
-            create=coordinator_create,
+            create=make_create_handler(  # lifted: shared body
+                coord_endpoint_config,
+                audit_emit=_audit_coordinator_create,
+            ),
             detail=coordinator_detail,
             open=make_open_handler(coord_endpoint_config),  # lifted: shared body
             close=make_close_handler(  # lifted: shared body

turnstone/core/attachments.py

@@ -14,8 +14,12 @@ from __future__ import annotations
 import collections
 import os
 import threading
+import uuid
 from dataclasses import dataclass
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from starlette.responses import JSONResponse
 
 # Byte caps — enforced by the server layer at upload time.  The
 # constants live here so the session / tests share the same definitions.
@@ -188,6 +192,104 @@ def classify_text_attachment(
     return "text/plain", None
 
 
+def validate_and_save_uploaded_files(
+    files: list[tuple[str, str, bytes]],
+    ws_id: str,
+    user_id: str,
+) -> tuple[list[str], JSONResponse | None]:
+    """Classify + save a list of ``(filename, claimed_mime, data)`` tuples.
+
+    Applies the same validation rules as ``upload_attachment`` (magic-byte
+    image sniffing, UTF-8 text decode, per-kind size cap, per-(ws,user)
+    pending cap) under the shared :func:`upload_lock`.
+
+    Kind-agnostic: both interactive and coordinator create-with-attachments
+    paths call into this helper from the lifted ``make_create_handler``
+    factory (Stage 2 ``create`` verb lift). The helper does not consult
+    any kind-specific config — the storage layer is kind-agnostic by
+    design (P1.5).
+
+    Returns ``(attachment_ids, None)`` on success or ``(ids_saved_so_far,
+    JSONResponse)`` on the first failure so the caller can roll back any
+    partial state.
+    """
+    from starlette.responses import JSONResponse as _JSONResponse
+
+    from turnstone.core.memory import list_pending_attachments, save_attachment
+
+    saved_ids: list[str] = []
+    if not files:
+        return saved_ids, None
+
+    lock = upload_lock(ws_id, user_id)
+    with lock:
+        pending_count = len(list_pending_attachments(ws_id, user_id))
+        for filename, claimed_mime, data in files:
+            if not data:
+                return saved_ids, _JSONResponse({"error": "Empty file"}, status_code=400)
+            sniffed_image = sniff_image_mime(data)
+            if sniffed_image is not None:
+                if len(data) > IMAGE_SIZE_CAP:
+                    return saved_ids, _JSONResponse(
+                        {
+                            "error": (
+                                f"Image too large ({len(data):,} bytes); "
+                                f"cap is {IMAGE_SIZE_CAP:,} bytes."
+                            ),
+                            "code": "too_large",
+                        },
+                        status_code=413,
+                    )
+                kind = "image"
+                mime = sniffed_image
+            else:
+                if len(data) > TEXT_DOC_SIZE_CAP:
+                    return saved_ids, _JSONResponse(
+                        {
+                            "error": (
+                                f"Text document too large ({len(data):,} bytes); "
+                                f"cap is {TEXT_DOC_SIZE_CAP:,} bytes."
+                            ),
+                            "code": "too_large",
+                        },
+                        status_code=413,
+                    )
+                mime_or_err = classify_text_attachment(filename, claimed_mime, data)
+                if mime_or_err[0] is None:
+                    return saved_ids, _JSONResponse(
+                        {"error": mime_or_err[1], "code": "unsupported"},
+                        status_code=400,
+                    )
+                kind = "text"
+                mime = mime_or_err[0]
+
+            if pending_count + 1 > MAX_PENDING_ATTACHMENTS_PER_USER_WS:
+                return saved_ids, _JSONResponse(
+                    {
+                        "error": (
+                            f"Too many pending attachments "
+                            f"(max {MAX_PENDING_ATTACHMENTS_PER_USER_WS} pending per workstream)"
+                        ),
+                        "code": "too_many",
+                    },
+                    status_code=409,
+                )
+            attachment_id = uuid.uuid4().hex
+            save_attachment(
+                attachment_id,
+                ws_id,
+                user_id,
+                filename,
+                mime,
+                len(data),
+                kind,
+                data,
+            )
+            saved_ids.append(attachment_id)
+            pending_count += 1
+    return saved_ids, None
+
+
 def unreadable_placeholder(filename: str) -> dict[str, Any]:
     """Return a content-part placeholder used when an attachment can't be
     decoded for a given turn.

turnstone/core/session_routes.py

@@ -148,6 +148,65 @@ class EventsReplay(Protocol):
 SseExecutorLookup = Callable[["Request"], Any]
 
 
+# (request, body, uid, uploaded_files) -> JSONResponse | None.
+# Optional kind-specific gate the lifted ``create`` body fires after
+# body parsing + uid resolution but before skill resolution and
+# ``mgr.create``. Returns ``None`` to continue, or a 4xx response
+# to short-circuit. Interactive wires gates for ws_id format,
+# kind=INTERACTIVE, parent_ws_id ownership, attachments+resume_ws
+# combo. Coord wires a 401-on-empty-uid (admin tokens always carry a
+# uid in practice; the gate is defensive). Mostly read-only — the
+# parent_ws_id ownership gate does a single storage lookup but
+# doesn't mutate anything.
+CreateRequestValidator = Callable[
+    ["Request", dict[str, Any], str, list[tuple[str, str, bytes]]],
+    "Awaitable[JSONResponse | None]",
+]
+# (request, body, uid, skill_data, skill_id, applied_skill_version) -> kwargs.
+# Builds the kwargs dict for ``mgr.create``. Both kinds call
+# ``mgr.create`` with the same callable shape, but the kwargs they
+# pass differ (interactive threads model + judge_model + client_type +
+# parent_ws_id + ws_id; coord threads only the smaller subset).
+# Captured in a per-kind callable rather than a flag-soup so the
+# kwargs dict construction stays readable at the wire-up site.
+CreateKwargsBuilder = Callable[
+    ["Request", dict[str, Any], str, dict[str, Any] | None, str, int],
+    dict[str, Any],
+]
+# (request, ws, body, uid, skill_data, applied_skill_version, attachment_ids) ->
+# extra response fields. Kind-specific tail end the lifted ``create``
+# body fires after the workstream is built, attachments are saved,
+# and audit is emitted. Returns extra fields to merge into the
+# response (e.g. interactive returns ``{resumed, message_count}``;
+# coord returns ``{}``). May spawn worker threads / register watch
+# runners / persist skill session config / dispatch initial messages
+# / pin routing. The factory does NOT wrap the call in try/except:
+# post-install failures should surface to the caller as 5xx so the
+# operator sees the misconfig instead of a half-built workstream.
+CreatePostInstall = Callable[
+    [
+        "Request",
+        "Workstream",
+        dict[str, Any],
+        str,
+        dict[str, Any] | None,
+        int,
+        list[str],
+    ],
+    "Awaitable[dict[str, Any]]",
+]
+# (request, ws, body, uid) -> None. Audit emitter for the create
+# event. Interactive emits ``workstream.created`` with
+# ``{kind, parent_ws_id}`` detail; coord emits ``coordinator.create``
+# with ``{coord_ws_id, src, name}`` detail. Wrapped in try/except by
+# the factory — audit-write failures shouldn't surface as HTTP 500
+# (mirrors the close / cancel / open lift contracts).
+CreateAuditEmitter = Callable[
+    ["Request", "Workstream", dict[str, Any], str],
+    None,
+]
+
+
 @dataclass(frozen=True)
 class AttachmentUploadHelpers:
     """Process-local hooks the lifted attachment factories call into.
@@ -282,6 +341,43 @@ class SessionEndpointConfig:
     # body falls through to the default executor. See
     # :data:`SseExecutorLookup` docstring above.
     sse_executor_lookup: SseExecutorLookup | None = None
+    # When ``True``, the lifted ``create`` body parses
+    # ``multipart/form-data`` (with one ``meta`` JSON field + zero or
+    # more ``file`` parts) in addition to plain ``application/json``.
+    # Both kinds wire ``True`` post-create-lift — coord gains
+    # create-time attachments here (§ Post-P3 reckoning item #1).
+    # The actual attachment validation+save+rollback always uses the
+    # storage layer (kind-agnostic since P1.5); this flag only
+    # toggles whether the multipart parse is attempted at all.
+    create_supports_attachments: bool = False
+    # When ``True``, the lifted ``create`` body honours a ``user_id``
+    # field in the request body if the caller's auth token comes from
+    # a trusted service (currently just ``"console"``). Interactive
+    # wires ``True`` so console-proxied creates can carry the real
+    # end user's identity through to the workstream owner. Coord
+    # wires ``False`` — coord create runs only on the console process
+    # and the operator's auth result is the source of truth.
+    create_supports_user_id_override: bool = False
+    # (request, body, uid, uploaded_files) -> JSONResponse | None.
+    # Per-kind pre-create gate (ws_id format, parent ownership, kind
+    # validation, etc. on interactive; 401-on-empty-uid on coord).
+    # ``None`` skips the gate entirely.
+    create_validate_request: CreateRequestValidator | None = None
+    # (request, body, uid, skill_data, skill_id, applied_skill_version)
+    # -> kwargs for ``mgr.create``. Required when the kind mounts a
+    # ``create`` handler — the lifted body has no opinion on the
+    # kind-specific kwarg shape and threads whatever this returns
+    # straight through to ``await asyncio.to_thread(mgr.create, **kwargs)``.
+    create_build_kwargs: CreateKwargsBuilder | None = None
+    # (request, ws, body, uid, skill_data, applied_skill_version,
+    # attachment_ids) -> extra response fields. Kind-specific tail
+    # end fired after attachments save + audit. Interactive returns
+    # ``{resumed, message_count}`` and spawns the initial-message
+    # worker thread; coord returns ``{}`` and dispatches via
+    # ``coord_adapter.send`` when an initial_message is provided.
+    # ``None`` skips the post-install entirely (response is just
+    # ``{ws_id, name, ...}`` with empty parity fields).
+    create_post_install: CreatePostInstall | None = None
 
 
 @dataclass(frozen=True)
@@ -1318,6 +1414,376 @@ def make_events_handler(cfg: SessionEndpointConfig) -> Handler:
     return events
 
 
+def make_create_handler(
+    cfg: SessionEndpointConfig,
+    *,
+    audit_emit: CreateAuditEmitter | None = None,
+) -> Handler:
+    """Lifted body for ``POST {prefix}/new`` — workstream creation.
+
+    Both kinds share the create sequence (parse body → resolve uid →
+    resolve skill → kind-specific validate → ``mgr.create`` → save
+    attachments → audit → kind-specific post-install → respond).
+    Per-kind divergence captured by the cfg + ``audit_emit``:
+
+    - ``cfg.create_supports_attachments`` — when ``True``, the body
+      may arrive as ``multipart/form-data`` with a ``meta`` JSON
+      field + ``file`` parts; uploads are validated post-create and
+      the workstream is rolled back if any file fails (interactive's
+      pre-lift pattern, lifted to coord here for parity).
+    - ``cfg.create_supports_user_id_override`` — when ``True``, a
+      ``user_id`` body field overrides the auth-derived uid if the
+      auth token is from a trusted service. Interactive ``True`` so
+      console-proxied creates carry the real end-user identity;
+      coord ``False``.
+    - ``cfg.create_validate_request`` — kind-specific pre-create
+      gates (interactive: ws_id format, kind, parent_ws_id ownership,
+      attachments+resume_ws combo; coord: 401-on-empty-uid).
+    - ``cfg.create_build_kwargs`` — kind-specific kwargs for
+      ``mgr.create``. Required when the kind mounts a create handler.
+    - ``cfg.create_post_install`` — kind-specific tail end (e.g.
+      interactive's resume + skill_config + initial-message worker
+      thread; coord's initial_message via coord_adapter.send).
+    - ``audit_emit`` — ``workstream.created`` on interactive,
+      ``coordinator.create`` on coord.
+
+    Behavior changes vs the pre-lift handlers (documented in
+    CHANGELOG, mostly coord-up-to-interactive parity gains):
+
+    - **Coord gains create-time attachments.** Pre-lift
+      ``coordinator_create`` accepted JSON only and ignored uploads;
+      the lifted body parses multipart bodies on coord and saves
+      attachments through the kind-agnostic storage layer (§ Post-P3
+      reckoning item #1). Coord's initial_message dispatch does NOT
+      yet reserve those attachments onto the first turn (coord
+      adapter's ``send`` doesn't take attachments) — the rows save
+      as pending and the first ``/send`` picks them up via the
+      standard send-with-attachments path. Initial-message + create-
+      time attachments coordination is a follow-up.
+    - **Coord gains the disabled-skill rejection.** Pre-lift
+      ``coordinator_create`` silently allowed disabled skills to
+      flow through to ``mgr.create``; the lifted body returns 400
+      ("Skill not found or disabled") matching interactive's
+      behaviour. Disabled skills are inert by definition; the gate
+      makes that explicit.
+    - **Both kinds converge on 200 OK.** Pre-lift interactive
+      returned 200 (default); coord returned 201. SDK consumers
+      that were branching on ``response.status == 201`` on coord
+      should switch to ``response.ok``. 200 was picked over 201 for
+      response-shape parity with the rest of the v1 surface (every
+      other shared verb returns 200), at the cost of leaving REST-
+      strictly-correct semantics on the table — a one-time release
+      note rather than ongoing client churn.
+    - **Both kinds converge on the manager-at-capacity 429
+      semantic.** Pre-lift interactive translated mgr.create's
+      ``RuntimeError`` to 400 ("invalid create request"); coord
+      already translated to 429. RuntimeError on ``SessionManager.create``
+      is documented as "manager at capacity" — 429 (rate-limit /
+      try-later) is the correct shape for both.
+    - **Both kinds converge on the factory-misconfig 503
+      semantic.** Pre-lift interactive let ``ValueError`` (raised by
+      the session factory on a misconfigured model alias) propagate
+      as 500; coord already translated to 503. The lifted body uses
+      503 with the factory's remediation text on both kinds —
+      operators get the actionable message instead of a generic
+      stack-traced 500.
+    - **Both kinds get a correlation_id'd 500 on unexpected
+      ``mgr.create`` failure.** Pre-lift interactive let unexpected
+      exceptions propagate as 500 with a stack-traced response
+      (potential information leak); coord already returned a
+      correlation_id'd 500 with the message redacted. The lifted
+      body adopts coord's safer pattern on both kinds.
+    - **Audit-emit failures no longer 500.** Pre-lift interactive
+      audit failures surfaced as HTTP 500 (no try/except); coord
+      swallowed via try/except + log.debug. The lifted body wraps
+      ``audit_emit`` in try/except + ``warning`` log, returning the
+      successful 201 to the caller. Mirrors the close / cancel /
+      open lift contracts.
+    - **Always-include response shape.** The lifted body always
+      returns ``{ws_id, name, resumed, message_count, attachment_ids}``,
+      with the parity fields defaulting to ``False`` / ``0`` / ``[]``
+      on kinds whose post-install doesn't populate them. SDK
+      consumers don't branch on kind.
+
+    Args:
+        cfg: per-kind policy bundle.
+        audit_emit: kind's audit emitter for the create event.
+            ``None`` skips the audit entirely.
+    """
+    # Lazy-imported at factory call time (mirrors the events lift) so
+    # ``session_routes.py``'s top-level import graph stays tight.
+
+    async def create(request: Request) -> Response:
+        import asyncio
+        import contextlib
+        import secrets
+
+        from turnstone.core.attachments import (
+            IMAGE_SIZE_CAP,
+            validate_and_save_uploaded_files,
+        )
+        from turnstone.core.web_helpers import (
+            read_json_or_400,
+            read_multipart_create_or_400,
+        )
+
+        if cfg.permission_gate is not None:
+            err = cfg.permission_gate(request)
+            if err is not None:
+                return err
+        mgr_opt, err503 = cfg.manager_lookup(request)
+        if err503 is not None:
+            return err503
+        # See ``make_approve_handler`` for the cast rationale.
+        mgr = cast("SessionManager", mgr_opt)
+
+        # --- Body parsing -------------------------------------------------
+        # Multipart only when the cfg lights up attachments AND the
+        # caller actually sent a multipart body. Plain JSON stays the
+        # default content type for both kinds.
+        content_type = (request.headers.get("content-type") or "").lower()
+        uploaded_files: list[tuple[str, str, bytes]] = []
+        body: dict[str, Any]
+        if cfg.create_supports_attachments and content_type.startswith("multipart/form-data"):
+            # Multipart cap: up to MAX_PENDING × image cap, plus slack
+            # for JSON meta + multipart framing. Per-file size is
+            # enforced inside :func:`validate_and_save_uploaded_files`
+            # against the kind-specific cap.
+            parsed = await read_multipart_create_or_400(
+                request,
+                max_files=10,
+                max_per_file_bytes=IMAGE_SIZE_CAP,
+                max_total_bytes=10 * IMAGE_SIZE_CAP,
+            )
+            if isinstance(parsed, JSONResponse):
+                return parsed
+            body, uploaded_files = parsed
+        else:
+            json_body = await read_json_or_400(request)
+            if isinstance(json_body, JSONResponse):
+                return json_body
+            body = json_body
+
+        # --- User id resolution ------------------------------------------
+        # Auth middleware populates ``request.state.auth_result`` for
+        # every authed request; we just read the user_id off it.
+        auth = getattr(getattr(request, "state", None), "auth_result", None)
+        uid: str = getattr(auth, "user_id", "") or ""
+        if cfg.create_supports_user_id_override:
+            # Trusted services (currently just ``console``) may forward
+            # the real end-user's id in the body so console-proxied
+            # creates carry the right owner. Token sources on end-user
+            # tokens (including console-proxy tokens that carry the
+            # real user's identity at the auth layer) are NOT trusted;
+            # only service identities. The deny-by-default keeps a
+            # malicious caller from impersonating other users.
+            body_uid = body.get("user_id")
+            if (
+                isinstance(body_uid, str)
+                and body_uid
+                and auth is not None
+                and getattr(auth, "token_source", "") in {"console"}
+            ):
+                uid = body_uid
+
+        # --- Per-kind pre-create validation ------------------------------
+        # Interactive validates ws_id format, kind, parent ownership,
+        # attachments+resume_ws combo. Coord 401s on empty uid.
+        if cfg.create_validate_request is not None:
+            err_validate = await cfg.create_validate_request(request, body, uid, uploaded_files)
+            if err_validate is not None:
+                return err_validate
+
+        # --- Skill resolution --------------------------------------------
+        # Both kinds resolve a body ``skill`` field through
+        # ``get_skill_by_name`` to the skill_data dict + the next
+        # applied_skill_version. Interactive previously skipped this
+        # entirely on resume_ws (the resumed session restores its own
+        # skill from config); the resume gate is captured by the
+        # interactive validator above (it returns 400 on the
+        # attachments+resume combo, but standalone resume_ws + skill
+        # is still allowed). To preserve that exact pre-lift skip on
+        # interactive, the validator may stash a sentinel — but
+        # simplest: re-read resume_ws_id here and skip skill lookup
+        # when both kinds see a non-empty resume_ws_id (coord doesn't
+        # support resume_ws today; the field is silently ignored).
+        # Strip whitespace on the skill name so a caller passing
+        # ``"skill": "  "`` is treated identically to ``"skill": ""``
+        # (skip skill resolution). Pre-lift coord explicitly stripped
+        # via ``(body.get("skill") or "").strip() or None``; pre-lift
+        # interactive didn't strip but never received whitespace-only
+        # skill names from the web UI. Convergence on the safer
+        # behaviour avoids a misleading 400 for an inert payload.
+        body_skill_raw = body.get("skill") or ""
+        body_skill = body_skill_raw.strip() if isinstance(body_skill_raw, str) else ""
+        resume_ws_id_raw = body.get("resume_ws") or ""
+        skill_data: dict[str, Any] | None = None
+        applied_skill_version = 0
+
+        # --- mgr.create (with skill resolution) -------------------------
+        # Skill lookup + version count + ``mgr.create`` all live inside
+        # one try/except so any storage failure during skill resolution
+        # gets the same correlation_id'd 500 as a ``mgr.create``
+        # exception. Pre-lift interactive let storage exceptions
+        # propagate to a stack-traced 500; the lifted body keeps the
+        # 500 status but redacts the message (operator gets the
+        # correlation id; logs carry the full ``exc_info``). The
+        # ``RuntimeError`` (capacity) and ``ValueError`` (factory
+        # misconfig) branches stay specific to ``mgr.create``: the
+        # skill-lookup path doesn't raise either of those.
+        if cfg.create_build_kwargs is None:
+            # The cfg required a build_kwargs callback for any kind
+            # mounting a create handler. Surface the misconfig as 500
+            # with a clear log line so the operator sees it instead of
+            # a confusing AttributeError.
+            log.error("ws.create.misconfigured_no_build_kwargs")
+            return JSONResponse(
+                {"error": "create handler misconfigured"},
+                status_code=500,
+            )
+        try:
+            if body_skill and not (isinstance(resume_ws_id_raw, str) and resume_ws_id_raw):
+                from turnstone.core.memory import get_skill_by_name
+                from turnstone.core.storage._registry import get_storage as _get_storage
+
+                skill_data = await asyncio.to_thread(get_skill_by_name, body_skill)
+                if not skill_data or not skill_data.get("enabled", False):
+                    return JSONResponse(
+                        {"error": f"Skill not found or disabled: {body_skill}"},
+                        status_code=400,
+                    )
+                tid = skill_data.get("template_id")
+                if tid:
+                    _st = _get_storage()
+                    if _st is not None:
+                        # ``count_skill_versions`` is best-effort: if the
+                        # version count call fails (transient storage
+                        # blip), default to 1 rather than aborting the
+                        # whole create. Persisted skill_version=1 is
+                        # the right semantic for the first applied
+                        # instance even if the count was unobtainable.
+                        try:
+                            applied_skill_version = (
+                                await asyncio.to_thread(_st.count_skill_versions, str(tid)) + 1
+                            )
+                        except Exception:
+                            log.debug(
+                                "ws.create.skill_version_failed skill=%s",
+                                body_skill,
+                                exc_info=True,
+                            )
+                            applied_skill_version = 1
+            skill_id_resolved = (
+                str(skill_data["template_id"])
+                if skill_data and skill_data.get("template_id")
+                else ""
+            )
+            kwargs = cfg.create_build_kwargs(
+                request, body, uid, skill_data, skill_id_resolved, applied_skill_version
+            )
+            ws = await asyncio.to_thread(mgr.create, **kwargs)
+        except RuntimeError as exc:
+            # ``SessionManager.create`` documents RuntimeError as
+            # "manager at capacity" — translate to 429 (rate-limit /
+            # try-later) on both kinds.
+            return JSONResponse({"error": str(exc)}, status_code=429)
+        except ValueError as exc:
+            # Session factory raises ValueError on misconfigured alias
+            # (model alias points at a model that no longer exists,
+            # etc.). Surface the factory's remediation text as 503 so
+            # operators get the actionable message instead of a
+            # stack-traced 500.
+            return JSONResponse({"error": str(exc)}, status_code=503)
+        except Exception:
+            # Don't echo the exception text — it can leak internal
+            # paths / frame names. Log with a correlation id and
+            # return that to the client so support can match a report
+            # to the log line.
+            correlation_id = secrets.token_hex(4)
+            log.warning(
+                "ws.create.failed correlation_id=%s",
+                correlation_id,
+                exc_info=True,
+            )
+            kind_noun = cfg.audit_action_prefix or "workstream"
+            return JSONResponse(
+                {
+                    "error": (
+                        f"failed to create {kind_noun} (internal error). "
+                        f"correlation_id={correlation_id}"
+                    )
+                },
+                status_code=500,
+            )
+
+        # --- Attachment validation + save + rollback --------------------
+        # Interactive's pre-lift pattern: validate post-create so
+        # ``ws_id`` is bound (the storage layer scopes attachments by
+        # ws_id). On any failure, mgr.close + delete_workstream the
+        # workstream so the caller doesn't see a half-built phantom.
+        # On coord this is a brand-new path — pre-lift coord_create
+        # had no attachments support. Note: coord's mgr.create already
+        # fired ``emit_created`` (cluster collector fan-out) by this
+        # point, so a rollback here produces a phantom create→close
+        # pair on the cluster events stream. Cluster consumers handle
+        # this gracefully (same shape as any quick-create-close), and
+        # the alternative (decoupling emit_created from mgr.create)
+        # is a bigger refactor that doesn't belong in the verb lift.
+        attachment_ids: list[str] = []
+        if uploaded_files:
+            saved_ids, save_err = await asyncio.to_thread(
+                validate_and_save_uploaded_files, uploaded_files, ws.id, uid
+            )
+            if save_err is not None:
+                from turnstone.core.memory import delete_workstream as _delete_ws
+
+                with contextlib.suppress(Exception):
+                    await asyncio.to_thread(mgr.close, ws.id)
+                with contextlib.suppress(Exception):
+                    await asyncio.to_thread(_delete_ws, ws.id)
+                return save_err
+            attachment_ids = saved_ids
+
+        # --- Audit emit --------------------------------------------------
+        if audit_emit is not None:
+            try:
+                audit_emit(request, ws, body, uid)
+            except Exception:
+                # Mirrors make_close_handler / make_cancel_handler /
+                # make_open_handler — audit-write failures shouldn't
+                # surface as HTTP 500. Log + continue.
+                log.warning(
+                    "ws.create.audit_failed ws=%s",
+                    ws.id[:8] if ws.id else "",
+                    exc_info=True,
+                )
+
+        # --- Per-kind post-install ---------------------------------------
+        extra_response: dict[str, Any] = {}
+        if cfg.create_post_install is not None:
+            extra_response = await cfg.create_post_install(
+                request,
+                ws,
+                body,
+                uid,
+                skill_data,
+                applied_skill_version,
+                attachment_ids,
+            )
+
+        return JSONResponse(
+            {
+                "ws_id": ws.id,
+                "name": ws.name,
+                "resumed": bool(extra_response.get("resumed", False)),
+                "message_count": int(extra_response.get("message_count", 0)),
+                "attachment_ids": attachment_ids,
+            }
+        )
+
+    return create
+
+
 def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
     """Lifted body for ``POST {prefix}/{ws_id}/send`` — message dispatch.
 

turnstone/server.py

@@ -66,6 +66,7 @@ from turnstone.core.session_routes import (
     make_attachment_handlers,
     make_cancel_handler,
     make_close_handler,
+    make_create_handler,
     make_dequeue_handler,
     make_events_handler,
     make_legacy_body_keyed_adapter,
@@ -2013,209 +2014,55 @@ def _reserve_and_resolve_attachments(
     return resolved, ordered_reserved, dropped
 
 
-def _validate_and_save_uploaded_files(
-    files: list[tuple[str, str, bytes]],
-    ws_id: str,
-    user_id: str,
-) -> tuple[list[str], JSONResponse | None]:
-    """Classify + save a list of ``(filename, claimed_mime, data)`` tuples.
+async def _interactive_create_validate_request(
+    request: Request,
+    body: dict[str, Any],
+    uid: str,
+    uploaded_files: list[tuple[str, str, bytes]],
+) -> JSONResponse | None:
+    """Per-kind pre-create gates for interactive workstreams.
 
-    Applies the same validation rules as ``upload_attachment`` (magic-byte
-    image sniffing, UTF-8 text decode, per-kind size cap, per-(ws,user)
-    pending cap) under the shared ``_attachment_upload_lock``.
+    Wired onto :attr:`SessionEndpointConfig.create_validate_request`
+    and called by :func:`make_create_handler` after body parsing
+    but before skill resolution / ``mgr.create``. Returns the
+    rejection response or ``None`` to continue.
 
-    Returns ``(attachment_ids, None)`` on success or ``(ids_saved_so_far,
-    JSONResponse)`` on the first failure so the caller can roll back any
-    partial state.
+    Gates:
+    - ws_id format must match :data:`_VALID_WS_ID` (32 hex chars)
+      when supplied.
+    - attachments + resume_ws combo is disallowed (resume forks an
+      existing ws; attachments belong on the *fresh* turn — caller
+      should resume first, then upload via the standard endpoint).
+    - body kind must be ``INTERACTIVE``: coordinator workstreams
+      land on the console handler with ``admin.coordinator`` scope,
+      not this one. Unknown / future kind values 400 rather than
+      silently coerce.
+    - parent_ws_id (when supplied) must reference a coordinator
+      owned by ``uid``. Without this gate an attacker could point a
+      new interactive workstream at someone else's coordinator and
+      receive that coordinator's child_ws_* SSE events
+      (name/state/tokens leak).
+    - notify_targets (when supplied) must validate.
+      :func:`_validate_notify_targets` is pure-read and doesn't need
+      ``ws`` to be built — gating here preserves pre-lift's 400
+      semantic for caller-supplied input. Without this pre-create
+      gate a malformed ``notify_targets`` would land in
+      ``post_install`` (after ``mgr.create``, audit emit, and the
+      ``ws_created`` broadcast) and the only available signal is to
+      raise — which the factory turns into 500. 400 at the gate is
+      correct shape for client-input validation.
     """
-    from turnstone.core.attachments import (
-        IMAGE_SIZE_CAP,
-        MAX_PENDING_ATTACHMENTS_PER_USER_WS,
-        TEXT_DOC_SIZE_CAP,
-        classify_text_attachment,
-        sniff_image_mime,
-        upload_lock,
-    )
-    from turnstone.core.memory import list_pending_attachments, save_attachment
-
-    saved_ids: list[str] = []
-    if not files:
-        return saved_ids, None
-
-    lock = upload_lock(ws_id, user_id)
-    with lock:
-        pending_count = len(list_pending_attachments(ws_id, user_id))
-        for filename, claimed_mime, data in files:
-            if not data:
-                return saved_ids, JSONResponse({"error": "Empty file"}, status_code=400)
-            sniffed_image = sniff_image_mime(data)
-            if sniffed_image is not None:
-                if len(data) > IMAGE_SIZE_CAP:
-                    return saved_ids, JSONResponse(
-                        {
-                            "error": (
-                                f"Image too large ({len(data):,} bytes); "
-                                f"cap is {IMAGE_SIZE_CAP:,} bytes."
-                            ),
-                            "code": "too_large",
-                        },
-                        status_code=413,
-                    )
-                kind = "image"
-                mime = sniffed_image
-            else:
-                if len(data) > TEXT_DOC_SIZE_CAP:
-                    return saved_ids, JSONResponse(
-                        {
-                            "error": (
-                                f"Text document too large ({len(data):,} bytes); "
-                                f"cap is {TEXT_DOC_SIZE_CAP:,} bytes."
-                            ),
-                            "code": "too_large",
-                        },
-                        status_code=413,
-                    )
-                mime_or_err = classify_text_attachment(filename, claimed_mime, data)
-                if mime_or_err[0] is None:
-                    return saved_ids, JSONResponse(
-                        {"error": mime_or_err[1], "code": "unsupported"},
-                        status_code=400,
-                    )
-                kind = "text"
-                mime = mime_or_err[0]
-
-            if pending_count + 1 > MAX_PENDING_ATTACHMENTS_PER_USER_WS:
-                return saved_ids, JSONResponse(
-                    {
-                        "error": (
-                            f"Too many pending attachments "
-                            f"(max {MAX_PENDING_ATTACHMENTS_PER_USER_WS} pending per workstream)"
-                        ),
-                        "code": "too_many",
-                    },
-                    status_code=409,
-                )
-            attachment_id = uuid.uuid4().hex
-            save_attachment(
-                attachment_id,
-                ws_id,
-                user_id,
-                filename,
-                mime,
-                len(data),
-                kind,
-                data,
-            )
-            saved_ids.append(attachment_id)
-            pending_count += 1
-    return saved_ids, None
-
-
-async def create_workstream(request: Request) -> JSONResponse:
-    """POST /v1/api/workstreams/new — create a new workstream.
-
-    Accepts two content types:
-
-    - ``application/json`` (default): body is a :class:`CreateWorkstreamRequest`.
-    - ``multipart/form-data``: one ``meta`` field (JSON object, same shape
-      as the JSON body) plus zero-or-more ``file`` parts. Files are saved
-      as attachments under the new workstream and reserved onto the first
-      ``initial_message`` turn (if provided) before the worker dispatches.
-    """
-    from turnstone.core.attachments import IMAGE_SIZE_CAP
-    from turnstone.core.audit import record_audit
-    from turnstone.core.memory import get_workstream_display_name
-    from turnstone.core.web_helpers import (
-        read_json_or_400,
-        read_multipart_create_or_400,
-    )
-
-    content_type = (request.headers.get("content-type") or "").lower()
-    uploaded_files: list[tuple[str, str, bytes]] = []
-    body: dict[str, Any]
-    if content_type.startswith("multipart/form-data"):
-        # Multipart cap: up to MAX_PENDING × image cap, plus slack for
-        # JSON meta + multipart framing.  Per-file size is enforced in
-        # _validate_and_save_uploaded_files against the kind-specific cap.
-        parsed = await read_multipart_create_or_400(
-            request,
-            max_files=10,
-            max_per_file_bytes=IMAGE_SIZE_CAP,
-            max_total_bytes=10 * IMAGE_SIZE_CAP,
-        )
-        if isinstance(parsed, JSONResponse):
-            return parsed
-        body, uploaded_files = parsed
-    else:
-        json_body = await read_json_or_400(request)
-        if isinstance(json_body, JSONResponse):
-            return json_body
-        body = json_body
-    mgr: SessionManager = request.app.state.workstreams
-    skip: bool = request.app.state.skip_permissions
-    auth = getattr(getattr(request, "state", None), "auth_result", None)
-    uid: str = getattr(auth, "user_id", "") or ""
-    # Trusted services (console) may forward the real user_id in the request
-    # body when creating workstreams on behalf of a user.  Only service
-    # identities are trusted — end-user tokens (including console-proxy tokens
-    # that carry the real user's identity) must not override user_id.
-    trusted_sources = {"console"}
-    if (
-        body.get("user_id")
-        and isinstance(body["user_id"], str)
-        and auth is not None
-        and auth.token_source in trusted_sources
-    ):
-        uid = body["user_id"]
-    body_skill = body.get("skill", "")
-    resume_ws_id = body.get("resume_ws", "")
-    # Resolve skill — applies content + session config (model, temperature, etc.)
-    # Skip when resuming: the resumed session restores its own skill from config.
-    skill_data: dict[str, Any] | None = None
-    if body_skill and not resume_ws_id:
-        from turnstone.core.memory import get_skill_by_name
-
-        skill_data = get_skill_by_name(body_skill)
-        if not skill_data or not skill_data.get("enabled", False):
-            return JSONResponse(
-                {"error": f"Skill not found or disabled: {body_skill}"},
-                status_code=400,
-            )
-    resolved_model = body.get("model") or None
-    if skill_data and skill_data.get("model"):
-        resolved_model = skill_data["model"]
-    resolved_skill: str | None = body_skill if skill_data else None
-    applied_skill_version = 0
-    if skill_data:
-        from turnstone.core.storage import get_storage as _get_storage
-
-        _st = _get_storage()
-        applied_skill_version = len(_st.list_skill_versions(skill_data["template_id"])) + 1
     requested_ws_id = body.get("ws_id", "") or ""
     if not isinstance(requested_ws_id, str):
         requested_ws_id = ""
     if requested_ws_id and not _VALID_WS_ID.match(requested_ws_id):
         return JSONResponse({"error": "invalid ws_id format"}, status_code=400)
-    # Disallow attachments + resume_ws in the same request — semantics
-    # are unclear (resume forks an existing ws, but attachments are for
-    # the *fresh* turn).  Caller should resume first, then upload via
-    # the standard endpoint.  Checked before mgr.create() so we don't
-    # waste work on a request we'll reject.
+    resume_ws_id = body.get("resume_ws", "") or ""
     if uploaded_files and resume_ws_id:
         return JSONResponse(
             {"error": "attachments cannot be combined with resume_ws"},
             status_code=400,
         )
-    # Kind + parent relationship: coordinator-spawned children forward
-    # these from the console's CoordinatorClient.  Default kind is
-    # ``INTERACTIVE``; coordinators themselves are created by the console's
-    # own SessionManager (coordinator kind) and never land in this handler.
-    # Only
-    # ``INTERACTIVE`` is accepted here — requests that try to create a
-    # coordinator via the generic workstream endpoint are rejected, and
-    # unknown kinds (typos, future-only values) are 400 rather than being
-    # silently coerced.  User-facing edge: the storage_edge and manager
-    # layers below re-validate, see comments at those sites for why.
     try:
         body_kind = WorkstreamKind.from_raw(body.get("kind"))
     except ValueError:
@@ -2235,16 +2082,6 @@ async def create_workstream(request: Request) -> JSONResponse:
         )
     body_parent = body.get("parent_ws_id") or None
     if body_parent is not None:
-        # Ownership gate: parent_ws_id is client-supplied in the request
-        # body, so a malicious caller could previously point a new
-        # interactive workstream at another tenant's coordinator — the
-        # coordinator's SSE fan-out would then route child_ws_* events
-        # (carrying the attacker's name/state/tokens) to that victim.
-        # Validate against storage: the parent must exist, be a
-        # coordinator, and belong to the same user.  Coordinator-spawned
-        # children satisfy this by construction (the coordinator's JWT
-        # `sub` claim is the owning user and parent_ws_id is the coord's
-        # own ws_id); external clients that fabricate the field get 403.
         from turnstone.core.storage._registry import get_storage as _get_storage_for_parent
 
         _pstorage = _get_storage_for_parent()
@@ -2262,251 +2099,295 @@ async def create_workstream(request: Request) -> JSONResponse:
                 {"error": "parent_ws_id must reference a coordinator you own"},
                 status_code=403,
             )
-    try:
-        ws = mgr.create(
-            user_id=uid,
-            name=body.get("name", ""),
-            model=resolved_model,
-            skill=resolved_skill,
-            skill_id=skill_data["template_id"] if skill_data else "",
-            skill_version=applied_skill_version,
-            ws_id=requested_ws_id,
-            client_type=body.get("client_type", "") or "",
-            judge_model=body.get("judge_model", "") or None,
-            parent_ws_id=body_parent,
-        )
-        if not isinstance(ws.ui, WebUI):
-            raise TypeError(f"Expected WebUI, got {type(ws.ui).__name__}")
-        if skip or body.get("auto_approve", False):
-            ws.ui.auto_approve = True
-        # Register watch runner for this workstream
-        runner = getattr(request.app.state, "watch_runner", None)
-        if runner and ws.session:
-            ws.session.set_watch_runner(
-                runner, dispatch_fn=_make_watch_dispatch(ws, ws.session, ws.ui)
-            )
-        gq: queue.Queue[dict[str, Any]] = request.app.state.global_queue
+    notify_targets_raw = body.get("notify_targets", "[]")
+    if isinstance(notify_targets_raw, list):
+        notify_targets_raw = json.dumps(notify_targets_raw)
+    _, nt_err = _validate_notify_targets(notify_targets_raw)
+    if nt_err:
+        return JSONResponse({"error": nt_err}, status_code=400)
+    return None
 
-        # Save attachments BEFORE the ws_created broadcast so failed
-        # validation doesn't make SSE consumers flash a workstream that
-        # never really existed.  Validate + save happens early; rollback
-        # is silent (no ws_created → no ws_closed needed).
-        attachment_ids: list[str] = []
-        if uploaded_files:
-            saved_ids, save_err = _validate_and_save_uploaded_files(uploaded_files, ws.id, uid)
-            if save_err is not None:
-                from turnstone.core.memory import delete_workstream as _delete_ws
 
-                with contextlib.suppress(Exception):
-                    mgr.close(ws.id)
-                with contextlib.suppress(Exception):
-                    _delete_ws(ws.id)
-                return save_err
-            attachment_ids = saved_ids
+def _interactive_create_build_kwargs(
+    request: Request,
+    body: dict[str, Any],
+    uid: str,
+    skill_data: dict[str, Any] | None,
+    skill_id: str,
+    applied_skill_version: int,
+) -> dict[str, Any]:
+    """Build kwargs for ``mgr.create`` from a parsed interactive create body.
 
-        # Emit creation event on global queue for SSE consumers (console).
-        # Deferred until past attachment validation so a rejected create
-        # doesn't surface a phantom create→close pair.
-        display_name = get_workstream_display_name(ws.id) or ws.name
-        with contextlib.suppress(queue.Full):
-            gq.put_nowait(
-                {
-                    "type": "ws_created",
-                    "ws_id": ws.id,
-                    "name": display_name,
-                    "model": ws.session.model if ws.session else "",
-                    "model_alias": ws.session.model_alias if ws.session else "",
-                    "kind": ws.kind,
-                    "parent_ws_id": ws.parent_ws_id,
-                    # Owner id is propagated through the cluster event
-                    # stream so console-side fan-out can enforce tenant
-                    # isolation — a coordinator must never receive
-                    # child_ws_* events for workstreams it doesn't own.
-                    "user_id": ws.user_id,
-                }
-            )
-        # Tamper-evident audit trail.  Lives alongside the broadcast
-        # event (not replacing it — the broadcast is ephemeral UI
-        # signalling, the audit row survives for forensic review).
-        _audit_storage = getattr(request.app.state, "auth_storage", None)
-        if _audit_storage is not None:
-            _, _audit_ip = _audit_context(request)
-            record_audit(
-                _audit_storage,
-                uid,
-                "workstream.created",
-                "workstream",
-                ws.id,
-                {"kind": str(ws.kind), "parent_ws_id": ws.parent_ws_id},
-                _audit_ip,
-            )
-        # Atomic workstream resume during creation.
-        resumed = False
-        message_count = 0
-        if resume_ws_id and ws.session is not None:
-            from turnstone.core.memory import resolve_workstream
+    Wired onto :attr:`SessionEndpointConfig.create_build_kwargs`. The
+    factory threads the resolved skill_data + skill_id + version
+    through; this builder picks the right model (skill override
+    beats body) and assembles the full kwargs dict that
+    ``SessionManager.create`` accepts (including the kind-specific
+    ``judge_model`` / ``client_type`` / ``parent_ws_id`` extras).
+    """
+    resolved_model = body.get("model") or None
+    if skill_data and skill_data.get("model"):
+        resolved_model = skill_data["model"]
+    requested_ws_id = body.get("ws_id", "") or ""
+    if not isinstance(requested_ws_id, str):
+        requested_ws_id = ""
+    return {
+        "user_id": uid,
+        "name": body.get("name", ""),
+        "model": resolved_model,
+        "skill": body.get("skill", "") if skill_data else None,
+        "skill_id": skill_id,
+        "skill_version": applied_skill_version,
+        "ws_id": requested_ws_id,
+        "client_type": body.get("client_type", "") or "",
+        "judge_model": body.get("judge_model", "") or None,
+        "parent_ws_id": body.get("parent_ws_id") or None,
+    }
 
-            target_id = resolve_workstream(resume_ws_id)
-            if target_id and ws.session.resume(target_id, fork=True):
-                resumed = True
-                message_count = len(ws.session.messages)
-                # If the user provided a custom name, set it as the fork's alias
-                # so it takes priority in display.  Otherwise keep the
-                # auto-generated name so auto-title can run fresh.
-                user_name = body.get("name", "").strip()
-                if user_name:
-                    from turnstone.core.memory import set_workstream_alias
 
-                    set_workstream_alias(ws.id, user_name)
-                    ws.name = user_name
-                ui = ws.ui
-                if isinstance(ui, WebUI):
-                    ui._enqueue({"type": "clear_ui"})
-                    history = _build_history(ws.session)
-                    if history:
-                        ui._enqueue({"type": "history", "messages": history})
-                # Broadcast a rename so the tab picks up the correct fork name
-                # (the ws_created event fired before fork with the pre-fork name).
-                with contextlib.suppress(queue.Full):
-                    gq.put_nowait({"type": "ws_rename", "ws_id": ws.id, "name": ws.name})
+async def _interactive_create_post_install(
+    request: Request,
+    ws: Workstream,
+    body: dict[str, Any],
+    uid: str,
+    skill_data: dict[str, Any] | None,
+    applied_skill_version: int,
+    attachment_ids: list[str],
+) -> dict[str, Any]:
+    """Tail end of interactive create: per-WebUI bookkeeping + dispatch.
 
-        # Apply skill session config (only for new workstreams with a skill)
-        if skill_data and not resumed and ws.session:
-            sess = ws.session
-            # Session settings from skill
-            if skill_data.get("temperature") is not None:
-                sess.temperature = skill_data["temperature"]
-            if skill_data.get("reasoning_effort"):
-                sess.reasoning_effort = skill_data["reasoning_effort"]
-            if skill_data.get("max_tokens") is not None:
-                sess.max_tokens = skill_data["max_tokens"]
-            if skill_data.get("token_budget", 0) > 0:
-                sess._token_budget = skill_data["token_budget"]
-            if skill_data.get("agent_max_turns") is not None:
-                sess.agent_max_turns = skill_data["agent_max_turns"]
-            # Approval policy
-            if skill_data.get("auto_approve"):
-                ws.ui.auto_approve = True
-            allowed = skill_data.get("allowed_tools", "")
-            if allowed and allowed != "[]":
-                # Parse as JSON array or comma-separated
-                import json as _json
+    Wired onto :attr:`SessionEndpointConfig.create_post_install`.
+    Runs after the workstream is fully built, attachments saved,
+    and audit emitted. Sequence:
 
-                try:
-                    tools_list = _json.loads(allowed)
-                except (ValueError, TypeError):
-                    tools_list = [t.strip() for t in allowed.split(",") if t.strip()]
-                if tools_list:
-                    ws.ui.auto_approve_tools = set(tools_list)
-            # Metadata
-            sess._notify_on_complete = skill_data.get("notify_on_complete", "{}")
-            sess._applied_skill_id = skill_data["template_id"]
-            sess._applied_skill_version = applied_skill_version
-            if skill_data.get("content"):
-                sess._applied_skill_content = skill_data["content"]
-            sess._save_config()
+    1. Cast ``ws.ui`` to :class:`WebUI` (defence in depth — the
+       interactive adapter's session factory is the only path that
+       reaches this handler).
+    2. Apply ``auto_approve`` from server-wide ``skip_permissions``
+       or per-request body.
+    3. Register the watch runner for the workstream's session.
+    4. Broadcast ``ws_created`` on the global SSE queue. Held until
+       this point so a rejected attachment validation produces no
+       phantom create→close pair on the SSE stream.
+    5. Atomic resume: if ``body["resume_ws"]`` is set, fork the
+       referenced session into the new ws_id, push history into the
+       UI listener queue, and rebroadcast ``ws_rename`` so the tab
+       picks up the fork's display name.
+    6. Apply the skill's session config (temperature / reasoning /
+       max_tokens / approval policy / metadata).
+    7. Resolve notify_targets (schedule targets win over skill
+       fallback).
+    8. Pin the workstream's routing to this node when no caller-
+       supplied ``ws_id`` was provided (direct creates).
+    9. Spawn the initial-message worker thread when ``initial_message``
+       is set, reserving any uploaded attachments for that first
+       turn.
 
-        # Resolve notify_targets: schedule targets override skill targets
-        notify_targets_raw = body.get("notify_targets", "[]")
-        if isinstance(notify_targets_raw, list):
-            notify_targets_raw = json.dumps(notify_targets_raw)
-        nt_str, nt_err = _validate_notify_targets(notify_targets_raw)
-        if nt_err:
-            return JSONResponse({"error": nt_err}, status_code=400)
-        # Skill fallback (only if schedule didn't specify targets)
-        if nt_str == "[]" and skill_data:
-            skill_notify = skill_data.get("notify_on_complete", "[]")
-            if skill_notify and skill_notify != "{}" and skill_notify != "[]":
-                fallback_str, fallback_err = _validate_notify_targets(skill_notify)
-                if not fallback_err:
-                    nt_str = fallback_str
-        ws.notify_targets = nt_str
+    Returns ``{resumed, message_count}`` for the response. On the
+    no-resume path both default to ``False`` / ``0``.
+    """
+    from turnstone.core.memory import get_workstream_display_name
 
-        # Pin locally-created workstreams so the console routes to this node.
-        # Console-routed creates pass ws_id in the request body — those are
-        # already bucket-aligned and don't need an override. Direct creates
-        # (web UI, watch, TurnstoneInit) generate their own ws_id, which may
-        # hash to a bucket assigned to a different node.
-        if not requested_ws_id:
-            node_id = getattr(request.app.state, "node_id", "")
-            if node_id:
-                try:
-                    from turnstone.core.storage import get_storage as _gs
-
-                    _gs().set_workstream_override(ws.id, node_id, reason="local")
-                except Exception:
-                    log.debug("Failed to set routing override for %s", ws.id, exc_info=True)
-
-        # If an initial_message was provided, send it as the first user message.
-        # This replaces the old bridge behavior where CreateWorkstreamMessage
-        # carried initial_message and the bridge sent it as a follow-up.
-        initial_message = body.get("initial_message", "").strip()
-        if initial_message and ws.session is not None:
-            session = ws.session
-            # Reserve any attachments uploaded in this request before the
-            # worker dispatches.  Mirrors the /send endpoint pattern: the
-            # send_id token scopes both the reservation and the eventual
-            # consume.  Unreserve on worker failure so the rows don't stay
-            # soft-locked forever.
-            send_id = uuid.uuid4().hex
-            resolved_atts: list[Any] = []
-            if attachment_ids:
-                resolved_atts, _ord, _drop = _reserve_and_resolve_attachments(
-                    attachment_ids, send_id, ws.id, uid
-                )
-
-            def _run_initial() -> None:
-                try:
-                    session.send(
-                        initial_message,
-                        attachments=resolved_atts or None,
-                        send_id=send_id if resolved_atts else None,
-                    )
-                except (Exception, GenerationCancelled):
-                    if attachment_ids:
-                        from turnstone.core.memory import (
-                            unreserve_attachments as _unreserve,
-                        )
-
-                        with contextlib.suppress(Exception):
-                            _unreserve(send_id, ws.id, uid)
-                    if isinstance(ws.ui, WebUI):
-                        ws.ui.on_stream_end()
-                        ws.ui.on_state_change("idle")
-                finally:
-                    try:
-                        last_content = _extract_last_assistant_content(session)
-                        _fire_notify_targets(ws, last_content)
-                    except Exception:
-                        log.warning("notify_completion.hook_error", ws_id=ws.id, exc_info=True)
-                    with ws._lock:
-                        ws._worker_running = False
-
-            # Inlined rather than via ``session_worker.send`` because
-            # at workstream creation no live worker can exist by
-            # construction — the enqueue branch of the shared dispatch
-            # is dead code here. We still set ``_worker_running`` and
-            # ``ws.worker_thread`` together under ``ws._lock`` so a
-            # /v1/api/send arriving immediately after creation observes
-            # the running state via the shared session_worker gate
-            # instead of racing into a parallel worker.
-            with ws._lock:
-                ws._worker_running = True
-                t = threading.Thread(target=_run_initial, daemon=True, name=f"ws-init-{ws.id[:8]}")
-                ws.worker_thread = t
-                t.start()
-
-        return JSONResponse(
+    if not isinstance(ws.ui, WebUI):
+        raise TypeError(f"Expected WebUI, got {type(ws.ui).__name__}")
+    skip: bool = request.app.state.skip_permissions
+    if skip or body.get("auto_approve", False):
+        ws.ui.auto_approve = True
+    runner = getattr(request.app.state, "watch_runner", None)
+    if runner and ws.session:
+        ws.session.set_watch_runner(runner, dispatch_fn=_make_watch_dispatch(ws, ws.session, ws.ui))
+    gq: queue.Queue[dict[str, Any]] = request.app.state.global_queue
+    # Emit ``ws_created`` on the global queue for SSE consumers
+    # (console). Held until past attachment validation in the
+    # factory so a rejected upload doesn't flash a workstream that
+    # never really existed.
+    display_name = get_workstream_display_name(ws.id) or ws.name
+    with contextlib.suppress(queue.Full):
+        gq.put_nowait(
             {
+                "type": "ws_created",
                 "ws_id": ws.id,
-                "name": ws.name,
-                "resumed": resumed,
-                "message_count": message_count,
-                "attachment_ids": attachment_ids,
+                "name": display_name,
+                "model": ws.session.model if ws.session else "",
+                "model_alias": ws.session.model_alias if ws.session else "",
+                "kind": ws.kind,
+                "parent_ws_id": ws.parent_ws_id,
+                # Owner id propagates through the cluster event
+                # stream so console-side fan-out can enforce tenant
+                # isolation — a coordinator must never receive
+                # child_ws_* events for workstreams it doesn't own.
+                "user_id": ws.user_id,
             }
         )
-    except RuntimeError as e:
-        return JSONResponse({"error": str(e)}, status_code=400)
+
+    # Atomic workstream resume during creation.
+    resumed = False
+    message_count = 0
+    resume_ws_id = body.get("resume_ws", "") or ""
+    if resume_ws_id and ws.session is not None:
+        from turnstone.core.memory import resolve_workstream
+
+        target_id = resolve_workstream(resume_ws_id)
+        if target_id and ws.session.resume(target_id, fork=True):
+            resumed = True
+            message_count = len(ws.session.messages)
+            user_name = body.get("name", "").strip()
+            if user_name:
+                from turnstone.core.memory import set_workstream_alias
+
+                set_workstream_alias(ws.id, user_name)
+                ws.name = user_name
+            ui = ws.ui
+            if isinstance(ui, WebUI):
+                ui._enqueue({"type": "clear_ui"})
+                history = _build_history(ws.session)
+                if history:
+                    ui._enqueue({"type": "history", "messages": history})
+            with contextlib.suppress(queue.Full):
+                gq.put_nowait({"type": "ws_rename", "ws_id": ws.id, "name": ws.name})
+
+    # Apply skill session config (only for new workstreams with a skill).
+    if skill_data and not resumed and ws.session:
+        sess = ws.session
+        if skill_data.get("temperature") is not None:
+            sess.temperature = skill_data["temperature"]
+        if skill_data.get("reasoning_effort"):
+            sess.reasoning_effort = skill_data["reasoning_effort"]
+        if skill_data.get("max_tokens") is not None:
+            sess.max_tokens = skill_data["max_tokens"]
+        if skill_data.get("token_budget", 0) > 0:
+            sess._token_budget = skill_data["token_budget"]
+        if skill_data.get("agent_max_turns") is not None:
+            sess.agent_max_turns = skill_data["agent_max_turns"]
+        if skill_data.get("auto_approve"):
+            ws.ui.auto_approve = True
+        allowed = skill_data.get("allowed_tools", "")
+        if allowed and allowed != "[]":
+            import json as _json
+
+            try:
+                tools_list = _json.loads(allowed)
+            except (ValueError, TypeError):
+                tools_list = [t.strip() for t in allowed.split(",") if t.strip()]
+            if tools_list:
+                ws.ui.auto_approve_tools = set(tools_list)
+        sess._notify_on_complete = skill_data.get("notify_on_complete", "{}")
+        sess._applied_skill_id = skill_data["template_id"]
+        sess._applied_skill_version = applied_skill_version
+        if skill_data.get("content"):
+            sess._applied_skill_content = skill_data["content"]
+        sess._save_config()
+
+    # notify_targets: schedule targets override skill targets. The
+    # validator already gated malformed input as 400; here we just
+    # canonicalise (list → JSON-encoded string) and apply the skill
+    # fallback if the caller didn't supply targets.
+    notify_targets_raw = body.get("notify_targets", "[]")
+    if isinstance(notify_targets_raw, list):
+        notify_targets_raw = json.dumps(notify_targets_raw)
+    nt_str, _ = _validate_notify_targets(notify_targets_raw)
+    if nt_str == "[]" and skill_data:
+        skill_notify = skill_data.get("notify_on_complete", "[]")
+        if skill_notify and skill_notify != "{}" and skill_notify != "[]":
+            fallback_str, fallback_err = _validate_notify_targets(skill_notify)
+            if not fallback_err:
+                nt_str = fallback_str
+    ws.notify_targets = nt_str
+
+    # Pin locally-created workstreams so the console routes to this node.
+    requested_ws_id = body.get("ws_id", "") or ""
+    if not requested_ws_id:
+        node_id = getattr(request.app.state, "node_id", "")
+        if node_id:
+            try:
+                from turnstone.core.storage import get_storage as _gs
+
+                _gs().set_workstream_override(ws.id, node_id, reason="local")
+            except Exception:
+                log.debug("Failed to set routing override for %s", ws.id, exc_info=True)
+
+    # Initial-message worker thread.
+    initial_message = body.get("initial_message", "").strip()
+    if initial_message and ws.session is not None:
+        session = ws.session
+        send_id = uuid.uuid4().hex
+        resolved_atts: list[Any] = []
+        if attachment_ids:
+            resolved_atts, _ord, _drop = _reserve_and_resolve_attachments(
+                attachment_ids, send_id, ws.id, uid
+            )
+
+        def _run_initial() -> None:
+            try:
+                session.send(
+                    initial_message,
+                    attachments=resolved_atts or None,
+                    send_id=send_id if resolved_atts else None,
+                )
+            except (Exception, GenerationCancelled):
+                if attachment_ids:
+                    from turnstone.core.memory import (
+                        unreserve_attachments as _unreserve,
+                    )
+
+                    with contextlib.suppress(Exception):
+                        _unreserve(send_id, ws.id, uid)
+                if isinstance(ws.ui, WebUI):
+                    ws.ui.on_stream_end()
+                    ws.ui.on_state_change("idle")
+            finally:
+                try:
+                    last_content = _extract_last_assistant_content(session)
+                    _fire_notify_targets(ws, last_content)
+                except Exception:
+                    log.warning("notify_completion.hook_error", ws_id=ws.id, exc_info=True)
+                with ws._lock:
+                    ws._worker_running = False
+
+        # Inlined rather than via ``session_worker.send`` because at
+        # workstream creation no live worker can exist by
+        # construction — the enqueue branch of the shared dispatch
+        # is dead code here. ``_worker_running`` + ``ws.worker_thread``
+        # are set together under ``ws._lock`` so a /v1/api/send
+        # arriving immediately after creation observes the running
+        # state via the shared session_worker gate instead of racing
+        # into a parallel worker.
+        with ws._lock:
+            ws._worker_running = True
+            t = threading.Thread(target=_run_initial, daemon=True, name=f"ws-init-{ws.id[:8]}")
+            ws.worker_thread = t
+            t.start()
+
+    return {"resumed": resumed, "message_count": message_count}
+
+
+def _audit_workstream_created(
+    request: Request,
+    ws: Workstream,
+    body: dict[str, Any],
+    uid: str,
+) -> None:
+    """Audit emitter for the interactive ``workstream.created`` event.
+
+    Wired onto :func:`make_create_handler` as ``audit_emit``. Runs
+    after the workstream is built and attachments saved; failures
+    are caught + logged by the factory (HTTP response stays 201).
+    """
+    from turnstone.core.audit import record_audit
+
+    _audit_storage = getattr(request.app.state, "auth_storage", None)
+    if _audit_storage is None:
+        return
+    _, _audit_ip = _audit_context(request)
+    record_audit(
+        _audit_storage,
+        uid,
+        "workstream.created",
+        "workstream",
+        ws.id,
+        {"kind": str(ws.kind), "parent_ws_id": ws.parent_ws_id},
+        _audit_ip,
+    )
 
 
 async def delete_workstream_endpoint(request: Request) -> JSONResponse:
@@ -3814,6 +3695,11 @@ def create_app(
         # the lifted contract — coord wires ``None`` and falls
         # back to the default executor.
         sse_executor_lookup=lambda request: request.app.state.sse_executor,
+        create_supports_attachments=True,
+        create_supports_user_id_override=True,
+        create_validate_request=_interactive_create_validate_request,
+        create_build_kwargs=_interactive_create_build_kwargs,
+        create_post_install=_interactive_create_post_install,
     )
     approve_handler = make_approve_handler(interactive_endpoint_config)
     close_handler = make_close_handler(
@@ -3830,6 +3716,10 @@ def create_app(
     send_handler = make_send_handler(interactive_endpoint_config)
     dequeue_handler = make_dequeue_handler(interactive_endpoint_config)
     attachment_handlers = make_attachment_handlers(interactive_endpoint_config)
+    create_handler = make_create_handler(
+        interactive_endpoint_config,
+        audit_emit=_audit_workstream_created,
+    )
     v1_routes: list[Any] = [
         Route("/api/events", make_legacy_query_keyed_adapter(events_handler)),
         Route("/api/events/global", global_events_sse),
@@ -3840,7 +3730,7 @@ def create_app(
         handlers=SharedSessionVerbHandlers(
             list_workstreams=list_workstreams,
             list_saved=list_saved_workstreams,
-            create=create_workstream,
+            create=create_handler,  # lifted: shared body
             close_legacy=make_legacy_body_keyed_adapter(close_handler),
             delete=delete_workstream_endpoint,
             open=open_handler,  # lifted: shared body