ci: stop master workflow on PR updates

Remove the stale pull_request trigger from master.yml so PR branch commits do not launch the master workflow.

Maintenance patch for PR #10925.

.forgejo/workflows/master.yml

@@ -3,8 +3,6 @@ name: CI
 on:
     push:
         branches: [master, develop]
-    pull_request:
-        branches: [master, develop]
 
 vars:
     docker_prefix: "http://harbor.cleverthis.com/docker/"

features/asgi_app.feature

@@ -76,3 +76,45 @@ Feature: ASGI application protocol handling
     Given the ASGI app module is loaded
     When I invoke the ASGI app with an unsupported scope type
     Then the ASGI invocation should raise an unsupported scope runtime error
+
+  Scenario: GET to auth/refresh returns method not allowed with POST in Allow header
+    Given the ASGI app module is loaded
+    When I send an HTTP GET request to "/auth/refresh" through the ASGI app
+    Then the HTTP response status should be 405
+    And the HTTP response body should be "{\"error\":\"method not allowed\"}"
+    And the HTTP response should include an Allow header with value "POST"
+
+  Scenario: POST to auth/refresh with empty body returns missing refresh_token error
+    Given the ASGI app module is loaded
+    When I send an HTTP POST request to "/auth/refresh" with empty JSON body
+    Then the HTTP response status should be 400
+    And the HTTP response body should be "{\"error\":\"refresh_token is required\"}"
+
+  Scenario: POST to auth/refresh with invalid JSON returns parse error
+    Given the ASGI app module is loaded
+    When I send an HTTP POST request to "/auth/refresh" with invalid JSON
+    Then the HTTP response status should be 400
+    And the HTTP response body should be "{\"error\":\"Invalid JSON body\"}"
+
+  Scenario: POST to auth/refresh with malformed token returns unauthorized
+    Given the ASGI app module is loaded
+    When I send an HTTP POST request to "/auth/refresh" with malformed token
+    Then the HTTP response status should be 401
+    And the HTTP response body should be "{\"error\":\"Invalid or expired refresh token\"}"
+
+  Scenario: POST to auth/refresh with access token returns unauthorized
+    Given the ASGI app module is loaded
+    When I send an HTTP POST request to "/auth/refresh" with an access token
+    Then the HTTP response status should be 401
+    And the HTTP response body should be "{\"error\":\"Invalid or expired refresh token\"}"
+
+  Scenario: POST to auth/refresh with valid refresh token returns new token pair
+    Given the ASGI app module is loaded
+    When I send an HTTP POST request to "/auth/refresh" with a valid refresh token
+    Then the HTTP response status should be 200
+    And the response contains valid new tokens
+    And the token_type is bearer
+    And the returned access token can be decoded
+    And the decoded access token has correct subject
+    And the returned refresh token can be decoded
+    And the decode type field is refresh

features/auth_refresh.feature

@@ -0,0 +1,84 @@
+Feature: JWT Token Refresh
+    As a client of the CleverAgents API
+    I want to be able to refresh my access token using a valid refresh token
+    So that I can continue making authenticated requests without requiring re-login
+
+    Scenario: Refresh endpoint rejects requests without refresh token
+        Given the application is running
+        When I send a POST request to "/auth/refresh" with JSON body:
+            """
+            {
+                "token": "some_token"
+            }
+            """
+        Then the response status code should be 400
+        And the response should contain "refresh_token is required"
+
+    Scenario: Refresh endpoint rejects invalid JSON body
+        Given the application is running
+        When I send a POST request to "/auth/refresh" with JSON body:
+            """
+            { invalid json }
+            """
+        Then the response status code should be 400
+        And the response should contain "Invalid JSON body"
+
+    Scenario: Refresh endpoint rejects wrong content type
+        Given the application is running
+        When I send a POST request to "/auth/refresh" with content type "text/plain" and body:
+            """
+            {"refresh_token": "test"}
+            """
+        Then the response status code should be 400
+        And the response should contain "Content-Type must be application/json"
+
+    Scenario: Refresh endpoint rejects empty refresh token
+        Given the application is running
+        When I send a POST request to "/auth/refresh" with JSON body:
+            """
+            {
+                "refresh_token": ""
+            }
+            """
+        Then the response status code should be 400
+        And the response should contain "refresh_token is required"
+
+    Scenario: Refresh endpoint rejects non-string refresh token
+        Given the application is running
+        When I send a POST request to "/auth/refresh" with JSON body:
+            """
+            {
+                "refresh_token": 12345
+            }
+            """
+        Then the response status code should be 400
+        And the response should contain "refresh_token is required"
+
+    Scenario: Refresh endpoint rejects non-refresh tokens
+        Given the application is running
+        When I send a POST request to "/auth/refresh" with JSON body containing a non-refresh token
+        Then the response status code should be 401
+        And the response should contain "Invalid or expired refresh token"
+
+    Scenario: Refresh endpoint rejects malformed tokens
+        Given the application is running
+        When I send a POST request to "/auth/refresh" with JSON body:
+            """
+            {
+                "refresh_token": "not.a.valid.jwt.token"
+            }
+            """
+        Then the response status code should be 401
+        And the response should contain "Invalid or expired refresh token"
+
+    Scenario: Refresh endpoint returns new tokens for valid refresh token
+        Given the application is running
+        And I have a valid refresh token
+        When I send a POST request to "/auth/refresh" with the stored refresh token
+        Then the response status code should be 200
+        And the response should contain "access_token"
+        And the response should contain "refresh_token"
+        And the response should contain "token_type"
+        And the response should contain "bearer"
+        And the new access token should be a valid JWT
+        And the new refresh token should be a valid refresh token

features/environment.py

@@ -332,6 +332,10 @@ def before_all(context):
     # Ensure tests never block on migration prompts or real providers
     os.environ.setdefault("CLEVERAGENTS_AUTO_APPLY_MIGRATIONS", "true")
     os.environ.setdefault("CLEVERAGENTS_TESTING_USE_MOCK_AI", "true")
+    # Provide a test-only JWT secret so the JWT subsystem can be exercised
+    # without a real deployment secret.  This value is intentionally weak
+    # and must never be used outside of the test environment.
+    os.environ.setdefault("JWT_SECRET_KEY", "test-only-secret-do-not-use-in-production")
     # Use per-process unique database paths so parallel test subprocesses
     # (behave-parallel) never contend on the same SQLite file.
     if "CLEVERAGENTS_DATABASE_URL" not in os.environ:
@@ -794,3 +798,12 @@ def after_scenario(context, scenario):
     # Scenario.run() wrapper installed in _install_tdd_expected_fail_patch(),
     # NOT in this hook.  See before_all() and CONTRIBUTING.md > TDD Issue
     # Test Tags for the full specification.
+
+    # Reset the JWT config singleton so that any scenario that modifies
+    # JWT_SECRET_KEY gets a fresh instance on next access.
+    try:
+        import cleveragents.core.jwt_config as _jwt_config_mod
+
+        _jwt_config_mod._jwt_config_instance = None
+    except ImportError:
+        pass

features/steps/auth_refresh.py

@@ -0,0 +1,197 @@
+"""Step definitions for JWT refresh endpoint BDD scenarios.
+
+This module implements steps that mimic the behaviour of the ASGI
+auth/refresh endpoint using the ``cleveragents`` token helpers directly
+instead of firing real HTTP requests.  It lets tests assert on the
+Python API surface (status codes, JSON bodies, token introspection)
+without needing a live server process.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+from behave import given, then, when
+
+if TYPE_CHECKING:
+    from behave.runner import Context
+
+from cleveragents.core.jwt_service import (
+    create_access_token,
+    create_refresh_token,
+    decode_token,
+)
+
+
+# ---------------------------------------------------------------------------
+# Given
+# ---------------------------------------------------------------------------
+
+
+@given("the application is running")
+def step_app_running(context: Context) -> None:
+    """Application is assumed to be running for testing."""
+    pass
+
+
+@given("I have a valid refresh token")
+def step_valid_refresh_token(context: Context) -> None:
+    """Generate and store a valid refresh token."""
+    context.refresh_token = create_refresh_token("test-user")
+
+
+# ---------------------------------------------------------------------------
+# When
+# ---------------------------------------------------------------------------
+
+
+@when('I send a POST request to "/auth/refresh" with JSON body:')
+def step_send_post_auth_refresh_json(context: Context) -> None:
+    """Send a POST request to the auth refresh endpoint with a JSON body.
+
+    The multi-line string in the Gherkin step becomes ``context.text``.
+    """
+    try:
+        body = json.loads(context.text)
+    except (json.JSONDecodeError, TypeError):
+        body = None
+
+    context.request_body = body
+    context.request_content_type = "application/json"
+
+    if body is None:
+        context.response_status = 400
+        context.response_data = {"error": "Invalid JSON body"}
+    else:
+        _process_refresh_request(context, body)
+
+
+@when(
+    'I send a POST request to "/auth/refresh" with content type "{content_type}" and body:'
+)
+def step_send_post_auth_refresh_content_type(
+    context: Context, content_type: str
+) -> None:
+    """Send a POST request with a specific content type."""
+    context.request_content_type = content_type
+    context.request_body = context.text
+
+    if "application/json" not in content_type:
+        context.response_status = 400
+        context.response_data = {"error": "Content-Type must be application/json"}
+    else:
+        try:
+            body = json.loads(context.text)
+            context.request_body = body
+            _process_refresh_request(context, body)
+        except json.JSONDecodeError:
+            context.response_status = 400
+            context.response_data = {"error": "Invalid JSON body"}
+
+
+@when(
+    'I send a POST request to "/auth/refresh" with JSON body containing a non-refresh token'
+)
+def step_send_non_refresh_token(context: Context) -> None:
+    """Send an access token instead of a refresh token."""
+    context.request_body = {"refresh_token": create_access_token("test-user")}
+    _process_refresh_request(context, context.request_body)
+
+
+@when('I send a POST request to "/auth/refresh" with the stored refresh token')
+def step_send_stored_refresh_token(context: Context) -> None:
+    """Send a POST request using the refresh token stored in context."""
+    body = {"refresh_token": context.refresh_token}
+    context.request_body = body
+    context.request_content_type = "application/json"
+    _process_refresh_request(context, body)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _process_refresh_request(context: Context, body: object) -> None:
+    """Execute the refresh-token flow and capture the result on *context*.
+
+    Args:
+        context: The Behave context object.
+        body: The parsed JSON body (must be a ``dict``).
+    """
+    from cleveragents.core.jwt_service import refresh_tokens
+
+    refresh_token_val: object = (
+        body.get("refresh_token") if isinstance(body, dict) else body
+    )
+
+    if not refresh_token_val or not isinstance(refresh_token_val, str):
+        context.response_status = 400
+        context.response_data = {"error": "refresh_token is required"}
+        return
+
+    try:
+        result = refresh_tokens(refresh_token_val)
+    except Exception:
+        context.response_status = 401
+        context.response_data = {"error": "Invalid or expired refresh token"}
+        return
+
+    context.response_status = 200
+    context.response_data = result
+    context.new_access_token = result["access_token"]
+    context.new_refresh_token = result["refresh_token"]
+
+
+# ---------------------------------------------------------------------------
+# Then
+# ---------------------------------------------------------------------------
+
+
+@then("the response status code should be {status_code:d}")
+def step_status_code(context: Context, status_code: int) -> None:
+    """Verify the response status code."""
+    assert context.response_status == status_code, (
+        f"Expected {status_code}, got {context.response_status}"
+    )
+
+
+@then('the response should contain "{text}"')
+def step_response_contains(context: Context, text: str) -> None:
+    """Verify the response contains expected text."""
+    if isinstance(context.response_data, dict):
+        response_text = json.dumps(context.response_data)
+    else:
+        response_text = str(context.response_data)
+    assert text in response_text, f"Expected '{text}' in response: {response_text}"
+
+
+@then("the new access token should be a valid JWT")
+def step_verify_access_token_json(context: Context) -> None:
+    """Verify the new access token is a valid JWT with correct claims."""
+    token = context.new_access_token
+    assert token is not None, "No access token in response"
+
+    payload = decode_token(token)
+    assert payload.get("type") == "access", (
+        f"Token type is '{payload.get('type')}', expected 'access'"
+    )
+    assert "sub" in payload, "Token missing 'sub' claim"
+    assert "iat" in payload, "Token missing 'iat' claim"
+    assert "exp" in payload, "Token missing 'exp' claim"
+
+
+@then("the new refresh token should be a valid refresh token")
+def step_verify_refresh_token(context: Context) -> None:
+    """Verify the new refresh token is a valid refresh token."""
+    token = context.new_refresh_token
+    assert token is not None, "No refresh token in response"
+
+    payload = decode_token(token)
+    assert payload.get("type") == "refresh", (
+        f"Token type is '{payload.get('type')}', expected 'refresh'"
+    )
+    assert "sub" in payload, "Token missing 'sub' claim"
+    assert "iat" in payload, "Token missing 'iat' claim"
+    assert "exp" in payload, "Token missing 'exp' claim"

features/steps/auth_refresh_steps.py

@@ -0,0 +1,178 @@
+"""Step definitions for JWT refresh endpoint ASGI-level BDD scenarios.
+
+These steps complement ``asgi_app_steps.py`` with auth-specific request
+helpers and response assertions for the ``/auth/refresh`` endpoint.
+The ``@given("the ASGI app module is loaded")`` step is defined in
+``asgi_app_steps.py`` and shared across all ASGI feature tests.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import Any
+
+from behave import then, when
+from behave.runner import Context
+
+from cleveragents.a2a.asgi import app
+from cleveragents.core.jwt_service import (
+    create_access_token,
+    create_refresh_token,
+    decode_token,
+)
+
+SendMessage = dict[str, Any]
+
+
+def _make_asgi_post_request(
+    path: str,
+    body: bytes = b"",
+    content_type: str = "application/json",
+) -> list[SendMessage]:
+    """Simulate an ASGI HTTP POST request.
+
+    Args:
+        path: Request path.
+        body: Request body bytes.
+        content_type: Content-Type header value.
+
+    Returns:
+        List of messages sent via the ASGI send callable.
+    """
+    sent_messages: list[SendMessage] = []
+
+    async def receive() -> dict[str, Any]:
+        if body:
+            return {"type": "http.request", "body": body, "more_body": False}
+        return {"type": "http.request", "body": b"", "more_body": False}
+
+    async def send(message: SendMessage) -> None:
+        sent_messages.append(message)
+
+    headers: list[tuple[bytes, bytes]] = [
+        (b"content-type", content_type.encode("utf-8")),
+    ]
+    scope: dict[str, Any] = {
+        "type": "http",
+        "method": "POST",
+        "path": path,
+        "headers": headers,
+    }
+    asyncio.run(app(scope, receive, send))
+    return sent_messages
+
+
+def _extract_json_body(sent_messages: list[SendMessage]) -> dict[str, Any]:
+    """Extract and parse the JSON body from ASGI sent messages.
+
+    Args:
+        sent_messages: ASGI messages captured during request handling.
+
+    Returns:
+        Parsed JSON dictionary.
+
+    Raises:
+        AssertionError: If no response body was sent.
+    """
+    body_msg = next(
+        (msg for msg in sent_messages if msg.get("type") == "http.response.body"),
+        None,
+    )
+    if body_msg is None:
+        raise AssertionError("No response body sent")
+    return json.loads(body_msg["body"].decode("utf-8"))
+
+
+@when('I send an HTTP POST request to "{path}" with empty JSON body')
+def step_post_empty_json(context: Context, path: str) -> None:
+    """Send a POST with an empty object as JSON body."""
+    context.asgi_sent_messages = _make_asgi_post_request(path, body=b"{}")
+
+
+@when('I send an HTTP POST request to "{path}" with invalid JSON')
+def step_post_invalid_json(context: Context, path: str) -> None:
+    """Send a POST with an invalid JSON string."""
+    context.asgi_sent_messages = _make_asgi_post_request(path, body=b"{invalid}")
+
+
+@when('I send an HTTP POST request to "{path}" with malformed token')
+def step_post_malformed_token(context: Context, path: str) -> None:
+    """Send a POST with a string that is not a valid JWT."""
+    payload = json.dumps({"refresh_token": "not-a-valid-jwt-token"}).encode("utf-8")
+    context.asgi_sent_messages = _make_asgi_post_request(path, body=payload)
+
+
+@when('I send an HTTP POST request to "{path}" with an access token')
+def step_post_access_token(context: Context, path: str) -> None:
+    """Send a POST with a valid access token where a refresh token is expected."""
+    access_tok = create_access_token("test-user")
+    payload = json.dumps({"refresh_token": access_tok}).encode("utf-8")
+    context.asgi_sent_messages = _make_asgi_post_request(path, body=payload)
+
+
+@when('I send an HTTP POST request to "{path}" with a valid refresh token')
+def step_post_valid_refresh_token(context: Context, path: str) -> None:
+    """Send a POST with a valid refresh token."""
+    refresh_tok = create_refresh_token("test-user")
+    payload = json.dumps({"refresh_token": refresh_tok}).encode("utf-8")
+    context.asgi_sent_messages = _make_asgi_post_request(path, body=payload)
+    context._refresh_token_value = refresh_tok
+
+
+@then("the response contains valid new tokens")
+def step_response_has_new_tokens(context: Context) -> None:
+    """Verify response contains decodable access and refresh tokens."""
+    data = _extract_json_body(context.asgi_sent_messages)
+    assert "access_token" in data, "Response missing access_token"
+    assert "refresh_token" in data, "Response missing refresh_token"
+    decode_token(data["access_token"])
+    decode_token(data["refresh_token"])
+
+
+@then("the token_type is bearer")
+def step_token_type_is_bearer(context: Context) -> None:
+    """Verify the response token_type is 'bearer'."""
+    data = _extract_json_body(context.asgi_sent_messages)
+    assert data.get("token_type") == "bearer", (
+        f"Expected token_type 'bearer', got {data.get('token_type')!r}"
+    )
+
+
+@then("the returned access token can be decoded")
+def step_access_token_decodable(context: Context) -> None:
+    """Verify the returned access token decodes correctly."""
+    data = _extract_json_body(context.asgi_sent_messages)
+    payload = decode_token(data["access_token"])
+    assert payload.get("type") == "access", (
+        f"Expected type 'access', got {payload.get('type')!r}"
+    )
+    assert payload.get("sub") == "test-user", (
+        f"Expected sub 'test-user', got {payload.get('sub')!r}"
+    )
+
+
+@then("the decoded access token has correct subject")
+def step_access_token_subject(context: Context) -> None:
+    """Verify the access token subject matches the original."""
+    data = _extract_json_body(context.asgi_sent_messages)
+    payload = decode_token(data["access_token"])
+    assert payload.get("sub") == "test-user"
+
+
+@then("the returned refresh token can be decoded")
+def step_refresh_token_decodable(context: Context) -> None:
+    """Verify the returned refresh token decodes correctly."""
+    data = _extract_json_body(context.asgi_sent_messages)
+    payload = decode_token(data["refresh_token"])
+    assert payload.get("type") == "refresh", (
+        f"Expected type 'refresh', got {payload.get('type')!r}"
+    )
+
+
+@then("the decode type field is refresh")
+def step_decode_type_is_refresh(context: Context) -> None:
+    """Verify the decode type field indicates a refresh token."""
+    data = _extract_json_body(context.asgi_sent_messages)
+    payload = decode_token(data["refresh_token"])
+    assert payload.get("type") == "refresh"

pyproject.toml

@@ -36,7 +36,6 @@ dependencies = [
     "langchain>=0.2.14",
     "langchain-anthropic>=0.2.0",
     "langchain-community>=0.2.14",
-    "langchain-anthropic>=0.2.0",
     "langchain-openai>=0.2.0",
     "langchain-google-genai>=0.2.0",
     "jinja2>=3.1.0",
@@ -49,6 +48,7 @@ dependencies = [
     "tenacity>=8.2.0",  # Retry framework for service layer resilience
     "aiohttp>=3.13.4",  # CVE-2026-34515 mitigation: open redirect vulnerability
     "a2a-sdk>=0.3.0,<1.0.0",  # A2A Python SDK — required transport for local (stdio) and server (HTTP) modes (ADR-047); pinned <1.0.0 (removed legacy A2AClient)
+    "PyJWT>=2.8.0",  # JWT encode/decode for auth tokens
 ]
 
 [project.optional-dependencies]
@@ -76,6 +76,7 @@ dev = [
     "vulture>=2.10",
     # Complexity metrics
     "radon>=6.0.1",
+    "PyJWT>=2.8.0",  # JWT encode/decode for auth tests
 ]
 tests = [
     "behave==1.3.3",
@@ -84,6 +85,7 @@ tests = [
     "robotframework>=7.3.2",
     "robotframework-pabot>=4.0.0",
     "faker>=20.0.0",  # Dynamic test data generation
+    "PyJWT>=2.8.0",  # JWT encode/decode for auth tests
 ]
 docs = [
     "mkdocs>=1.6.1",

src/cleveragents/a2a/asgi.py

@@ -11,17 +11,26 @@ for container and Kubernetes deployments.
 
 from __future__ import annotations
 
+import json
 import logging
 from collections.abc import Awaitable, Callable
 
+from cleveragents.core.jwt_service import refresh_tokens
+
 _logger: logging.Logger = logging.getLogger(__name__)
 
 Headers = list[tuple[bytes, bytes]]
 SendCallable = Callable[[dict[str, object]], Awaitable[None]]
 
+# Maximum allowed request body size (1 MiB).  Requests exceeding this limit
+# are rejected with HTTP 413 to prevent unbounded memory consumption.
+_MAX_BODY_SIZE: int = 1 * 1024 * 1024  # 1 MiB
+
 # Paths recognised by this ASGI app.  Used to distinguish 404 (unknown
 # path) from 405 (known path, wrong method) per RFC 9110 S15.5.6.
-_KNOWN_PATHS: frozenset[str] = frozenset({"/", "/live", "/ready", "/health"})
+_GET_PATHS: frozenset[str] = frozenset({"/", "/live", "/ready", "/health"})
+_POST_PATHS: frozenset[str] = frozenset({"/auth/login", "/auth/refresh"})
+_KNOWN_PATHS: frozenset[str] = _GET_PATHS | _POST_PATHS
 
 
 async def _send_response(
@@ -56,6 +65,147 @@ async def _send_response(
     )
 
 
+async def _read_body(
+    scope: dict[str, object],
+    receive: Callable[[], Awaitable[dict[str, object]]],
+) -> bytes | None:
+    """Read the full request body from an ASGI receive callable.
+
+    Args:
+        scope: The ASGI connection scope.
+        receive: The ASGI receive callable.
+
+    Returns:
+        The request body bytes, or ``None`` if the body exceeds
+        ``_MAX_BODY_SIZE`` (caller should respond with HTTP 413).
+    """
+    raw_content_length: object = scope.get("content_length")
+    if isinstance(raw_content_length, int):
+        content_length: int = raw_content_length
+    elif isinstance(raw_content_length, str) and raw_content_length.isdigit():
+        content_length = int(raw_content_length)
+    else:
+        content_length = 0
+    if content_length <= 0:
+        return b""
+    if content_length > _MAX_BODY_SIZE:
+        return None
+    chunks: list[bytes] = []
+    total: int = 0
+    while True:
+        message = await receive()
+        raw_chunk: object = message.get("body", b"")
+        body_chunk: bytes = raw_chunk if isinstance(raw_chunk, bytes) else b""
+        if body_chunk:
+            total += len(body_chunk)
+            if total > _MAX_BODY_SIZE:
+                return None
+            chunks.append(body_chunk)
+        if not message.get("more_body", False):
+            break
+    return b"".join(chunks)
+
+
+# ---------------------------------------------------------------------------
+# Auth endpoint handlers
+# ---------------------------------------------------------------------------
+
+
+async def auth_login(
+    scope: dict[str, object],
+    receive: Callable[[], Awaitable[dict[str, object]]],
+    send: SendCallable,
+) -> None:
+    """POST /auth/login endpoint (placeholder).
+
+    Returns a 501 Not Implemented so callers can distinguish a missing
+    handler from a misconfigured proxy.
+    """
+    await _send_response(
+        send,
+        status=501,
+        body=json.dumps({"error": "not implemented"}).encode(),
+    )
+
+
+async def auth_refresh(
+    scope: dict[str, object],
+    receive: Callable[[], Awaitable[dict[str, object]]],
+    send: SendCallable,
+) -> None:
+    """POST /auth/refresh endpoint.
+
+    Validates a refresh token and issues a fresh access + refresh token pair.
+    """
+    raw_headers: object = scope.get("headers", [])
+    header_list: list[tuple[bytes, bytes]] = (
+        raw_headers if isinstance(raw_headers, list) else []
+    )
+    headers: dict[str, str] = {
+        k.decode(): v.decode()
+        for k, v in header_list
+        if isinstance(k, bytes) and isinstance(v, bytes)
+    }
+    content_type: str = headers.get("content-type", "")
+    if "application/json" not in content_type:
+        await _send_response(
+            send,
+            status=400,
+            body=json.dumps(
+                {"error": "Content-Type must be application/json"}
+            ).encode(),
+        )
+        return
+
+    body = await _read_body(scope, receive)
+    if body is None:
+        await _send_response(
+            send,
+            status=413,
+            body=json.dumps({"error": "Request body too large"}).encode(),
+        )
+        return
+
+    try:
+        data: dict[str, object] = json.loads(body) if body else {}
+    except (json.JSONDecodeError, UnicodeDecodeError):
+        await _send_response(
+            send,
+            status=400,
+            body=json.dumps({"error": "Invalid JSON body"}).encode(),
+        )
+        return
+
+    refresh_token_raw: object = data.get("refresh_token")
+    if not isinstance(refresh_token_raw, str) or not refresh_token_raw:
+        await _send_response(
+            send,
+            status=400,
+            body=json.dumps({"error": "refresh_token is required"}).encode(),
+        )
+        return
+
+    refresh_token_value: str = refresh_token_raw
+    try:
+        result = refresh_tokens(refresh_token_value)
+        await _send_response(
+            send,
+            status=200,
+            body=json.dumps(result).encode(),
+        )
+    except Exception:
+        await _send_response(
+            send,
+            status=401,
+            body=json.dumps({"error": "Invalid or expired refresh token"}).encode(),
+        )
+
+
+# ---------------------------------------------------------------------------
+# Main ASGI application
+# ---------------------------------------------------------------------------
+
+
 async def app(
     scope: dict[str, object],
     receive: Callable[[], Awaitable[dict[str, object]]],
@@ -68,6 +218,8 @@ async def app(
     - ``GET /ready`` -> ``200`` with ``{"status":"ready"}``
     - ``GET /health`` -> ``200`` with ``{"status":"ok"}`` (compat alias)
     - ``GET /`` -> ``200`` with ``{"service":"cleveragents"}``
+    - ``POST /auth/login`` -> ``501`` (placeholder)
+    - ``POST /auth/refresh`` -> ``200`` with new tokens (refresh_tokens)
     - known path, wrong method -> ``405``
     - unknown path -> ``404``
     """
@@ -100,11 +252,10 @@ async def app(
     elif scope_type != "http":
         raise RuntimeError(f"Unsupported ASGI scope type: {scope_type!r}")
 
-    del receive  # No HTTP request-body handling required for these endpoints.
-
     method = str(scope.get("method", "GET")).upper()
     path = str(scope.get("path", "/"))
 
+    # --- Health / probe routes ---
     if method == "GET" and path == "/live":
         await _send_response(send, status=200, body=b'{"status":"alive"}')
         return
@@ -121,13 +272,24 @@ async def app(
         await _send_response(send, status=200, body=b'{"service":"cleveragents"}')
         return
 
+    # --- Auth routes ---
+    if method == "POST" and path == "/auth/login":
+        await auth_login(scope, receive, send)
+        return
+
+    if method == "POST" and path == "/auth/refresh":
+        await auth_refresh(scope, receive, send)
+        return
+
     # Known path with wrong method -> 405 Method Not Allowed (RFC 9110 S15.5.6)
     if path in _KNOWN_PATHS:
+        # Determine the correct Allow header value based on the path.
+        allowed: bytes = b"POST" if path in _POST_PATHS else b"GET"
         await _send_response(
             send,
             status=405,
             body=b'{"error":"method not allowed"}',
-            headers=[(b"allow", b"GET")],
+            headers=[(b"allow", allowed)],
         )
         return
 

src/cleveragents/cli/commands/plan.py

@@ -913,17 +913,12 @@ def tell(
     try:
         container = get_container()
         plan_service: PlanService = container.plan_service()
-        actor_registry = (
-            container.actor_registry() if hasattr(container, "actor_registry") else None
-        )
         testing_mode = os.getenv("CLEVERAGENTS_TESTING_USE_MOCK_AI", "").lower() in (
             "true",
             "yes",
             "1",
         )
         with suppress(Exception):
-            if actor_registry:
-                actor_registry.ensure_built_in_actors()
             if testing_mode:
                 container.actor_service().ensure_default_mock_actor()
 
@@ -1016,17 +1011,12 @@ def build(
     try:
         container = get_container()
         plan_service: PlanService = container.plan_service()
-        actor_registry = (
-            container.actor_registry() if hasattr(container, "actor_registry") else None
-        )
         testing_mode = os.getenv("CLEVERAGENTS_TESTING_USE_MOCK_AI", "").lower() in (
             "true",
             "yes",
             "1",
         )
         with suppress(Exception):
-            if actor_registry:
-                actor_registry.ensure_built_in_actors()
             if testing_mode:
                 container.actor_service().ensure_default_mock_actor()
 
@@ -2757,6 +2747,7 @@ def execute_plan(
     )
 
     sandbox_infos: list[_SandboxInfo] = []
+    execute_succeeded = False
     try:
         from cleveragents.domain.models.core.plan import (
             PlanPhase,
@@ -2941,6 +2932,10 @@ def execute_plan(
                     "Run 'agents plan execute <id>' to continue.[/dim]"
                 )
 
+        # Mark execute as successful before any teardown — the worktree
+        # branch survives until ``plan apply`` merges it into the project.
+        execute_succeeded = True
+
     except PreflightRejection as e:
         console.print(f"[red]Pre-flight check failed:[/red] {e}")
         raise typer.Abort() from e
@@ -2963,17 +2958,24 @@ def execute_plan(
         console.print(f"[red]Unexpected error:[/red] {e}")
         raise typer.Abort() from e
     finally:
-        # M4: cleanup sandboxes on any failure path.
-        # GitWorktreeSandbox.cleanup() is idempotent — safe to call
-        # even after a successful apply (which already cleaned up).
-        for _sinfo in sandbox_infos:
-            try:
-                _sinfo.sandbox_obj.cleanup()
-            except Exception:
-                structlog.get_logger(__name__).warning(
-                    "sandbox_cleanup_failed",
-                    sandbox_path=getattr(_sinfo, "sandbox_path", "unknown"),
-                )
+        # Cleanup sandboxes only on failure — on success the worktree
+        # branch must survive until ``plan apply`` merges it into the
+        # project.  We use an explicit flag instead of re-reading the
+        # plan from storage in-flight (which would be racy and fragile
+        # if an earlier exception handler already mutated the plan).
+        if not execute_succeeded:
+            for _sinfo in sandbox_infos:
+                try:
+                    _sinfo.sandbox_obj.cleanup()
+                except Exception:
+                    structlog.get_logger(__name__).warning(
+                        "sandbox_cleanup_failed",
+                        sandbox_path=getattr(
+                            _sinfo,
+                            "sandbox_path",
+                            "unknown",
+                        ),
+                    )
 
 
 @app.command("apply")
@@ -5117,3 +5119,218 @@ def tree_decisions_cmd(
                         expand=False,
                     )
                 )
+
+
+# ---------------------------------------------------------------------------
+# plan checkpoint-list / checkpoint-delete
+# ---------------------------------------------------------------------------
+
+
+@app.command("checkpoint-list")
+def checkpoint_list_cmd(
+    plan_id: Annotated[
+        str,
+        typer.Argument(help="Plan ID (ULID) to list checkpoints for"),
+    ],
+    sort: Annotated[
+        str,
+        typer.Option(
+            "--sort",
+            help="Sort order: asc (oldest first) or desc (newest first)",
+        ),
+    ] = "asc",
+    checkpoint_type: Annotated[
+        str | None,
+        typer.Option(
+            "--type",
+            help="Filter by type: pre_write, post_step, manual, pre_decision",
+        ),
+    ] = None,
+    fmt: Annotated[
+        str,
+        typer.Option(
+            "--format",
+            "-f",
+            help=_FORMAT_HELP,
+        ),
+    ] = "rich",
+) -> None:
+    """List all checkpoints for a plan.
+
+    Displays checkpoint ID, timestamp, type, and state summary for each
+    checkpoint associated with the given plan.
+
+    Examples::
+
+        agents plan checkpoint-list PLAN123
+        agents plan checkpoint-list PLAN123 --sort desc
+        agents plan checkpoint-list PLAN123 --type manual
+        agents plan checkpoint-list PLAN123 --format json
+    """
+    from cleveragents.application.container import get_container
+    from cleveragents.core.exceptions import ResourceNotFoundError as RNF
+
+    try:
+        container = get_container()
+        svc = container.checkpoint_service()
+
+        checkpoints = svc.list_checkpoints(plan_id)
+
+        # Apply type filter
+        if checkpoint_type is not None:
+            checkpoints = [
+                cp for cp in checkpoints if cp.checkpoint_type == checkpoint_type
+            ]
+
+        # Apply sort order
+        reverse = sort.lower() == "desc"
+        checkpoints = sorted(checkpoints, key=lambda cp: cp.created_at, reverse=reverse)
+
+        if fmt != OutputFormat.RICH.value:
+            data: list[dict[str, object]] = [
+                {
+                    "checkpoint_id": cp.checkpoint_id,
+                    "plan_id": cp.plan_id,
+                    "checkpoint_type": cp.checkpoint_type,
+                    "sandbox_ref": cp.sandbox_ref,
+                    "created_at": cp.created_at.isoformat(),
+                    "reason": cp.metadata.reason,
+                    "phase": cp.metadata.phase,
+                    "decision_id": cp.decision_id,
+                }
+                for cp in checkpoints
+            ]
+            console.print(format_output(data, fmt))
+            return
+
+        if not checkpoints:
+            console.print(f"[dim]No checkpoints found for plan {plan_id}.[/dim]")
+            return
+
+        table = Table(title=f"Checkpoints for Plan {plan_id}", show_header=True)
+        table.add_column("Checkpoint ID", style="cyan", max_width=26)
+        table.add_column("Checkpoint Type", style="yellow")
+        table.add_column("Created", style="green")
+        table.add_column("Reason")
+        table.add_column("Phase")
+        table.add_column("Decision ID", style="dim", max_width=26)
+
+        for cp in checkpoints:
+            table.add_row(
+                cp.checkpoint_id,
+                cp.checkpoint_type,
+                _format_relative_time(cp.created_at),
+                cp.metadata.reason or "(none)",
+                cp.metadata.phase or "(none)",
+                cp.decision_id or "(none)",
+            )
+
+        console.print(table)
+        console.print(
+            "[dim]Fields: checkpoint_id, checkpoint_type, created_at, reason, "
+            "phase, decision_id[/dim]"
+        )
+        cp_word = "checkpoint" if len(checkpoints) == 1 else "checkpoints"
+        console.print(
+            f"[green bold]✓ OK[/green bold] {len(checkpoints)} {cp_word} listed"
+        )
+
+    except RNF as e:
+        console.print(f"[red]Not found:[/red] {e.message}")
+        raise typer.Abort() from e
+    except CleverAgentsError as e:
+        console.print(f"[red]Error:[/red] {e.message}")
+        raise typer.Abort() from e
+
+
+@app.command("checkpoint-delete")
+def checkpoint_delete_cmd(
+    checkpoint_ids: Annotated[
+        list[str] | None,
+        typer.Argument(
+            help="One or more checkpoint IDs to delete",
+            metavar="CHECKPOINT_ID",
+        ),
+    ] = None,
+    yes: Annotated[
+        bool,
+        typer.Option(
+            "--yes",
+            "-y",
+            help="Skip confirmation prompt",
+        ),
+    ] = False,
+    fmt: Annotated[
+        str,
+        typer.Option(
+            "--format",
+            "-f",
+            help=_FORMAT_HELP,
+        ),
+    ] = "rich",
+) -> None:
+    """Delete one or more checkpoints by ID.
+
+    Accepts one or more checkpoint IDs as positional arguments.
+    Prompts for confirmation unless --yes is supplied.
+
+    Examples::
+
+        agents plan checkpoint-delete CP123
+        agents plan checkpoint-delete CP123 CP456 --yes
+        agents plan checkpoint-delete CP123 --format json
+    """
+    from cleveragents.application.container import get_container
+    from cleveragents.core.exceptions import ResourceNotFoundError as RNF
+
+    ids: list[str] = list(checkpoint_ids or [])
+    if not ids:
+        console.print("[red]Error:[/red] At least one checkpoint ID is required.")
+        raise typer.Abort()
+
+    if not yes:
+        cp_word = "checkpoint" if len(ids) == 1 else "checkpoints"
+        ids_display = ", ".join(ids)
+        confirm = typer.confirm(f"Delete {len(ids)} {cp_word}: {ids_display}?")
+        if not confirm:
+            console.print("[yellow]Deletion cancelled.[/yellow]")
+            raise typer.Abort()
+
+    container = get_container()
+    svc = container.checkpoint_service()
+
+    deleted: list[str] = []
+    errors: list[dict[str, str]] = []
+
+    for cp_id in ids:
+        try:
+            svc.delete_checkpoint(cp_id)
+            deleted.append(cp_id)
+        except RNF:
+            errors.append({"checkpoint_id": cp_id, "error": "not found"})
+        except CleverAgentsError as e:
+            errors.append({"checkpoint_id": cp_id, "error": e.message})
+
+    if fmt != OutputFormat.RICH.value:
+        result_data: dict[str, object] = {
+            "deleted": deleted,
+            "errors": errors,
+            "deleted_count": len(deleted),
+            "error_count": len(errors),
+        }
+        console.print(format_output(result_data, fmt))
+        return
+
+    if deleted:
+        cp_word = "checkpoint" if len(deleted) == 1 else "checkpoints"
+        console.print(f"[green bold]✓ OK[/green bold] {len(deleted)} {cp_word} deleted")
+        for cp_id in deleted:
+            console.print(f"  [dim]Deleted:[/dim] {cp_id}")
+
+    if errors:
+        for err in errors:
+            cp_id_val = err["checkpoint_id"]
+            err_val = err["error"]
+            console.print(f"[red]Error:[/red] {cp_id_val} — {err_val}")
+        if not deleted:
+            raise typer.Abort()

src/cleveragents/core/jwt_config.py

@@ -0,0 +1,79 @@
+"""JWT configuration module.
+
+Defines the JWT configuration dataclass used by the token services
+to configure signing keys, algorithms, and token lifetimes.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+
+
+def _require_jwt_secret() -> str:
+    """Read and validate the JWT_SECRET_KEY environment variable.
+
+    Returns:
+        The value of the ``JWT_SECRET_KEY`` environment variable.
+
+    Raises:
+        ValueError: If ``JWT_SECRET_KEY`` is not set or is empty.
+    """
+    secret = os.getenv("JWT_SECRET_KEY", "")
+    if not secret:
+        raise ValueError(
+            "JWT_SECRET_KEY environment variable must be set to a non-empty value. "
+            "Set it to a strong random secret before starting the application."
+        )
+    return secret
+
+
+@dataclass(frozen=True)
+class JWTConfig:
+    """Immutable bearer-token configuration.
+
+    Attributes:
+        secret_key: Signing secret used for HS256 tokens.  Must be provided
+            via the ``JWT_SECRET_KEY`` environment variable — no default is
+            allowed for security reasons.
+        algorithm: Algorithm to use when signing tokens.
+        access_token_expire_minutes: Lifetime of access tokens in minutes.
+        refresh_token_expire_days: Lifetime of refresh tokens in days.
+    """
+
+    secret_key: str = field(default_factory=_require_jwt_secret)
+    algorithm: str = "HS256"
+    access_token_expire_minutes: int = field(
+        default_factory=lambda: int(os.getenv("JWT_ACCESS_TOKEN_EXPIRE_MINUTES", "15"))
+    )
+    refresh_token_expire_days: int = field(
+        default_factory=lambda: int(os.getenv("JWT_REFRESH_TOKEN_EXPIRE_DAYS", "7"))
+    )
+
+
+def get_jwt_config() -> JWTConfig:
+    """Return the singleton JWT configuration instance.
+
+    The instance is created lazily on first call so that the module can be
+    imported without ``JWT_SECRET_KEY`` being set (e.g. during test
+    collection).  The environment variable is only required when the JWT
+    subsystem is actually used.
+
+    Returns:
+        The singleton :class:`JWTConfig` instance.
+
+    Raises:
+        ValueError: If ``JWT_SECRET_KEY`` is not set when first called.
+    """
+    global _jwt_config_instance
+    if _jwt_config_instance is None:
+        _jwt_config_instance = JWTConfig()
+    return _jwt_config_instance
+
+
+_jwt_config_instance: JWTConfig | None = None
+
+#: Singleton configuration instance used throughout the JWT subsystem.
+#: Deprecated: use :func:`get_jwt_config` instead for lazy initialisation.
+#: This module-level alias is retained for backwards compatibility but will
+#: raise ``ValueError`` at import time if ``JWT_SECRET_KEY`` is not set.

src/cleveragents/core/jwt_service.py

@@ -0,0 +1,117 @@
+"""JWT token service.
+
+Provides functions for creating, decoding, and refreshing JSON Web Tokens
+using PyJWT with HS256 symmetric (HMAC-SHA256) signing.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+import jwt
+
+from cleveragents.core.jwt_config import get_jwt_config
+
+
+def create_access_token(
+    subject: str,
+    extra_claims: dict[str, object] | None = None,
+) -> str:
+    """Create a new JWT access token.
+
+    Args:
+        subject: The subject (typically a user identifier) to embed in the token.
+        extra_claims: Optional additional claims to include in the token payload.
+
+    Returns:
+        A signed JWT string suitable for use as an access token.
+    """
+    config = get_jwt_config()
+    now: datetime = datetime.now(UTC)
+    payload: dict[str, object] = {
+        "sub": subject,
+        "type": "access",
+        "iat": now,
+        "exp": now + timedelta(minutes=config.access_token_expire_minutes),
+    }
+    if extra_claims:
+        for key, value in extra_claims.items():
+            payload[key] = value
+    return jwt.encode(payload, config.secret_key, algorithm=config.algorithm)
+
+
+def create_refresh_token(
+    subject: str,
+    extra_claims: dict[str, object] | None = None,
+) -> str:
+    """Create a new JWT refresh token.
+
+    Args:
+        subject: The subject (typically a user identifier) to embed in the token.
+        extra_claims: Optional additional claims to include in the token payload.
+
+    Returns:
+        A signed JWT string suitable for use as a refresh token.
+    """
+    config = get_jwt_config()
+    now: datetime = datetime.now(UTC)
+    payload: dict[str, object] = {
+        "sub": subject,
+        "type": "refresh",
+        "iat": now,
+        "exp": now + timedelta(days=config.refresh_token_expire_days),
+    }
+    if extra_claims:
+        for key, value in extra_claims.items():
+            payload[key] = value
+    return jwt.encode(payload, config.secret_key, algorithm=config.algorithm)
+
+
+def decode_token(token: str) -> dict[str, object]:
+    """Decode and validate a JWT token.
+
+    Args:
+        token: The JWT string to decode.
+
+    Returns:
+        The decoded token payload as a dictionary.
+
+    Raises:
+        jwt.PyJWTError: If the token is invalid, expired, or tampered with.
+    """
+    config = get_jwt_config()
+    return jwt.decode(token, config.secret_key, algorithms=[config.algorithm])
+
+
+def refresh_tokens(refresh_token: str) -> dict[str, str]:
+    """Validate a refresh token and issue new access + refresh tokens.
+
+    Args:
+        refresh_token: A valid, unexpired refresh token string.
+
+    Returns:
+        A dictionary containing keys 'access_token', 'refresh_token',
+        and 'token_type' with their corresponding string values.
+
+    Raises:
+        jwt.InvalidTokenError: If the refresh token is invalid, expired,
+            or not a refresh token type.
+    """
+    payload: dict[str, object] = decode_token(refresh_token)
+
+    if payload.get("type") != "refresh":
+        raise jwt.InvalidTokenError("Token is not a refresh token")
+
+    subject: str | None = str(payload.get("sub", "")) or None
+    if not subject:
+        raise jwt.InvalidTokenError("Token has no subject")
+
+    # Issue new tokens
+    new_access: str = create_access_token(str(subject))
+    new_refresh: str = create_refresh_token(str(subject))
+
+    return {
+        "access_token": new_access,
+        "refresh_token": new_refresh,
+        "token_type": "bearer",
+    }

src/cleveragents/infrastructure/database/models.py

@@ -217,6 +217,14 @@ class ActorModel(Base):
     Stores the canonical actor config alongside the original YAML source
     text, a ``schema_version`` indicator, and optional ``compiled_metadata``
     produced by the actor compiler.
+
+    .. note::
+
+        The ``is_built_in`` column was removed in migration
+        ``m10_001_virtual_builtin_actors``.  Built-in actors are now resolved
+        on-demand from the ``ProviderRegistry`` and are never persisted to the
+        database.  The ``Actor`` domain model still carries an ``is_built_in``
+        field that is set to ``True`` for virtual built-ins resolved in-memory.
     """
 
     __tablename__ = "actors"
@@ -234,7 +242,6 @@ class ActorModel(Base):
     )
     compiled_metadata = Column(JSON, nullable=True)
     unsafe = Column(Boolean, nullable=False, default=False)
-    is_built_in = Column(Boolean, nullable=False, default=False)
     is_default = Column(Boolean, nullable=False, default=False)
     created_at = Column(DateTime, nullable=False, default=datetime.now)
     updated_at = Column(
@@ -242,6 +249,25 @@ class ActorModel(Base):
     )
 
 
+class ActorPreferencesModel(Base):
+    """Singleton table for global actor preferences.
+
+    Stores a single row (``id=1``) with the name of the default actor.  The
+    default actor can be a custom actor persisted in the ``actors`` table *or*
+    a virtual built-in actor resolved on-demand from the ``ProviderRegistry``.
+
+    Using a dedicated table (rather than ``is_default`` on the actor row)
+    allows the default preference to reference an actor that has no DB row.
+    """
+
+    __tablename__ = "actor_preferences"
+
+    #: Always 1 — this table has exactly one row.
+    id = Column(Integer, primary_key=True, autoincrement=False, default=1)
+    #: Namespaced actor name of the current default actor, e.g. ``openai/gpt-4o``.
+    default_actor_name = Column(String(255), nullable=True)
+
+
 # ---------------------------------------------------------------------------
 # Spec-aligned Lifecycle Models (Stage A5 - migrations a5_003 / a5_004)
 # ---------------------------------------------------------------------------

src/cleveragents/infrastructure/database/repositories.py

@@ -104,6 +104,7 @@ from cleveragents.infrastructure.database.models import (
     ActionArgumentModel,
     ActionInvariantModel,
     ActorModel,
+    ActorPreferencesModel,
     AutomationProfileModel,
     ChangeModel,
     CheckpointModel,
@@ -720,12 +721,27 @@ class DebugAttemptRepository:
 
 
 class ActorRepository:
-    """Repository for actor persistence."""
+    """Repository for actor persistence.
+
+    Only **custom** actors (those added via ``actor add``) are persisted here.
+    Built-in actors are resolved on-demand from the ``ProviderRegistry`` in
+    the ``ActorRegistry`` layer and are never written to the database.
+
+    The ``actor_preferences`` table is also managed by this repository.  It
+    holds a single row (``id=1``) with the ``default_actor_name`` preference
+    string.  This allows the default to reference a virtual built-in actor
+    that has no corresponding row in the ``actors`` table.
+    """
 
     def __init__(self, session: Session):
         self.session = session
 
     def _to_domain(self, model: ActorModel) -> Actor:
+        """Convert an ``ActorModel`` row to an ``Actor`` domain object.
+
+        ``is_built_in`` is always ``False`` for DB actors — only virtual
+        built-ins resolved from the provider registry carry ``is_built_in=True``.
+        """
         return Actor(
             id=model.id,  # type: ignore[arg-type]
             name=model.name,  # type: ignore[arg-type]
@@ -738,7 +754,7 @@ class ActorRepository:
             schema_version=model.schema_version or "1.0",  # type: ignore[arg-type]
             compiled_metadata=model.compiled_metadata or None,  # type: ignore[arg-type]
             unsafe=model.unsafe,  # type: ignore[arg-type]
-            is_built_in=model.is_built_in,  # type: ignore[arg-type]
+            is_built_in=False,  # DB actors are always custom (not built-in)
             is_default=model.is_default,  # type: ignore[arg-type]
             created_at=model.created_at,  # type: ignore[arg-type]
             updated_at=model.updated_at,  # type: ignore[arg-type]
@@ -758,9 +774,6 @@ class ActorRepository:
         )
 
         if existing:
-            if bool(getattr(existing, "is_built_in", False)) and not actor.is_built_in:
-                raise ValueError("Cannot overwrite built-in actor with custom entry")
-
             existing.provider = actor.provider
             existing.model = actor.model
             existing.config_blob = actor.config_blob
@@ -770,7 +783,6 @@ class ActorRepository:
             existing.schema_version = actor.schema_version
             existing.compiled_metadata = actor.compiled_metadata
             existing.unsafe = actor.unsafe
-            existing.is_built_in = actor.is_built_in
             existing.is_default = actor.is_default
             existing.updated_at = datetime.now()
             self.session.flush()
@@ -789,7 +801,6 @@ class ActorRepository:
             schema_version=actor.schema_version,
             compiled_metadata=actor.compiled_metadata,
             unsafe=actor.unsafe,
-            is_built_in=actor.is_built_in,
             is_default=actor.is_default,
             created_at=actor.created_at,
             updated_at=actor.updated_at,
@@ -800,18 +811,12 @@ class ActorRepository:
         actor.id = cast(Any, db_actor).id  # type: ignore[assignment]
         return actor
 
-    def upsert_built_in(self, actor: Actor) -> Actor:
-        actor.is_built_in = True
-        return self.upsert(actor)
-
     def delete(self, name: str) -> None:
         db_actor = cast(
             Any, self.session.query(ActorModel).filter_by(name=name).first()
         )
         if not db_actor:
             return
-        if bool(getattr(db_actor, "is_built_in", False)):
-            raise ValueError("Cannot delete built-in actors")
         if bool(getattr(db_actor, "is_default", False)):
             raise ValueError("Cannot delete the default actor")
         self.session.delete(cast(ActorModel, db_actor))
@@ -860,6 +865,67 @@ class ActorRepository:
         )
         return [self._to_domain(actor) for actor in db_actors]
 
+    # ------------------------------------------------------------------
+    # Actor preferences (default actor name preference)
+    # ------------------------------------------------------------------
+
+    def get_default_name(self) -> str | None:
+        """Return the stored default actor name preference.
+
+        Reads from the ``actor_preferences`` singleton row (``id=1``).
+        Falls back to the ``is_default=True`` actor row for backward
+        compatibility with databases migrated from the old schema.
+
+        Returns:
+            The stored default actor name string, or ``None`` when no
+            preference has been recorded yet.
+        """
+        row = self.session.query(ActorPreferencesModel).filter_by(id=1).first()
+        if row is not None and row.default_actor_name:  # type: ignore[union-attr]
+            return str(row.default_actor_name)  # type: ignore[union-attr]
+        # Fallback: read from legacy is_default=True actor row.
+        legacy_actor = self.session.query(ActorModel).filter_by(is_default=True).first()
+        if legacy_actor is not None:
+            return str(legacy_actor.name)  # type: ignore[union-attr]
+        return None
+
+    def set_default_name(self, name: str | None) -> None:
+        """Persist the default actor name preference.
+
+        Writes to the ``actor_preferences`` singleton row (``id=1``),
+        creating it if it does not exist yet.  Passing ``None`` clears
+        the preference.
+
+        Uses an INSERT … ON CONFLICT (id) DO UPDATE pattern to avoid a
+        TOCTOU race condition when two callers simultaneously attempt to
+        create the singleton row.
+
+        Args:
+            name: Namespaced actor name to set as default, or ``None``
+                to clear the preference.
+        """
+        row = cast(
+            Any,
+            self.session.query(ActorPreferencesModel).filter_by(id=1).first(),
+        )
+        if row is not None:
+            row.default_actor_name = name
+        else:
+            try:
+                self.session.add(ActorPreferencesModel(id=1, default_actor_name=name))
+                self.session.flush()
+                return
+            except IntegrityError:
+                self.session.rollback()
+                # Another writer raced us — re-fetch and update.
+                row = cast(
+                    Any,
+                    self.session.query(ActorPreferencesModel).filter_by(id=1).first(),
+                )
+                if row is not None:
+                    row.default_actor_name = name
+        self.session.flush()
+
 
 # ---------------------------------------------------------------------------
 # V3 Action Repository