DO NOT MERGE — merge train B: 3468,3467,3437,3478,3477,3339,3486,3412

docker/base-image/agent_server/models.py

@@ -1,7 +1,7 @@
 """
 Pydantic models for the agent server.
 """
-from pydantic import BaseModel
+from pydantic import BaseModel, SecretStr
 from typing import List, Optional, Dict, Any
 from datetime import datetime
 
@@ -16,6 +16,45 @@ class ChatMessage(BaseModel):
     timestamp: datetime = None
 
 
+class AuthOverride(BaseModel):
+    """A credential for THIS ONE spawn (#3470 — SUB-003 walks the pool).
+
+    Exactly one of the two is set. The backend sends it when it re-issues a
+    refused turn on another credential: the agent applies it as the top env
+    layer of that single `claude` subprocess and changes nothing in the
+    container, so concurrent turns may run on different credentials and a
+    plain `docker restart` cannot desynchronise anything. Claude Code only;
+    other runtimes ignore it (they authenticate from their own `.env`).
+    """
+    # SecretStr: a `repr()` of the request, a validation error's echoed
+    # `input`, or a stray `logger.info(f"{request}")` prints `**********`,
+    # never the credential. Read with `.get_secret_value()`.
+    oauth_token: Optional[SecretStr] = None  # CLAUDE_CODE_OAUTH_TOKEN for this spawn
+    api_key: Optional[SecretStr] = None      # ANTHROPIC_API_KEY for this spawn
+
+    @staticmethod
+    def _value(secret: Optional[SecretStr]) -> Optional[str]:
+        return secret.get_secret_value() if secret is not None else None
+
+    def kind(self) -> Optional[str]:
+        """``"oauth_token"`` / ``"api_key"`` / None when neither is usable."""
+        token, key = self._value(self.oauth_token), self._value(self.api_key)
+        if token and not key:
+            return "oauth_token"
+        if key and not token:
+            return "api_key"
+        return None
+
+    def secret(self) -> Optional[str]:
+        """The one usable credential's plaintext, or None."""
+        kind = self.kind()
+        if kind == "oauth_token":
+            return self._value(self.oauth_token)
+        if kind == "api_key":
+            return self._value(self.api_key)
+        return None
+
+
 class ChatRequest(BaseModel):
     message: str
     stream: bool = False
@@ -26,6 +65,8 @@ class ChatRequest(BaseModel):
     # fresh session that is NOT kept as the agent's chat session, so the next
     # caller never resumes a context with the skill loaded. Claude Code only.
     isolated_session: bool = False
+    # #3470: per-spawn credential for a SUB-003 re-issue. Ignored by old images.
+    auth_override: Optional[AuthOverride] = None
 
 
 class ChatResponse(BaseModel):
@@ -255,6 +296,8 @@ class ParallelTaskRequest(BaseModel):
     # result-callback endpoint. Ignored by non-Claude runtimes / old images
     # (they run synchronously and return 200 — the backend's non-202 fallback).
     async_result: Optional[bool] = False
+    # #3470: per-spawn credential for a SUB-003 re-issue (see AuthOverride).
+    auth_override: Optional[AuthOverride] = None
 
 
 class ParallelTaskResponse(BaseModel):

docker/base-image/agent_server/routers/chat.py

@@ -8,21 +8,53 @@
 import logging
 from datetime import datetime
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Response
 from fastapi.responses import StreamingResponse, JSONResponse
 
 from ..models import ChatRequest, ModelRequest, ParallelTaskRequest
 from ..state import agent_state
 from ..services.claude_code import get_execution_lock
+from ..services.execution_env import AUTH_OVERRIDE_HEADER
 from ..services.runtime_adapter import get_runtime
 from ..services.process_registry import get_process_registry, PENDING_CHAT_TIMEOUT_SECONDS
 from ..services import result_callback
 from ..services import retained_results
+from ..utils.credential_sanitizer import add_credential_value
 
 logger = logging.getLogger(__name__)
 router = APIRouter()
 
 
+def _accept_auth_override(request, runtime):
+    """The request's per-spawn credential (#3470), or None when this runtime
+    cannot honour one. Stages the value for log redaction. Never logs it."""
+    override = getattr(request, "auth_override", None)
+    if override is None or override.kind() is None:
+        return None
+    if not getattr(runtime, "supports_auth_override", False):
+        logger.info("[#3470] auth_override ignored: runtime does not take a Claude credential")
+        return None
+    add_credential_value(override.secret())
+    return override
+
+
+def _stamp_auth_override(response: Response, override) -> None:
+    """Tell the backend the override WAS applied. A 429 body carries no
+    metadata, so the header is the only signal that works on the failure path
+    too; an older image never sets it, which is how the backend tells the two
+    apart (`subscription_auto_switch.AUTH_OVERRIDE_HEADER`)."""
+    # `response` is None only for a direct in-process call (tests); FastAPI
+    # injects it for every HTTP request regardless of the default.
+    if override is not None and response is not None:
+        response.headers[AUTH_OVERRIDE_HEADER] = override.kind()
+
+
+def _reraise_with_auth_override(exc: HTTPException, override) -> None:
+    if override is not None:
+        exc.headers = {**(exc.headers or {}), AUTH_OVERRIDE_HEADER: override.kind()}
+    raise exc
+
+
 def _retain_terminal(execution_id, envelope: dict) -> None:
     """#2944: keep this turn's terminal where the backend watchdog can claim it
     if the connection that was waiting for it is gone (a backend recreate
@@ -36,7 +68,7 @@ def _retain_terminal(execution_id, envelope: dict) -> None:
 
 
 @router.post("/api/chat")
-async def chat(request: ChatRequest):
+async def chat(request: ChatRequest, response: Response = None):
     """
     Send a message to Claude Code and get response with execution log.
 
@@ -72,6 +104,9 @@ async def chat(request: ChatRequest):
             runtime = get_runtime()
             # Use request.model if provided, otherwise use the model set via /api/model endpoint
             effective_model = request.model or agent_state.current_model
+            # #3470: a SUB-003 re-issue's per-spawn credential (Claude only).
+            auth_override = _accept_auth_override(request, runtime)
+            execute_kwargs = {"auth_override": auth_override} if auth_override is not None else {}
             # #1020: feed the richer /health signal — count this execution and
             # record success/failure (drives consecutive_failures).
             agent_state.record_task_start()
@@ -84,6 +119,7 @@ async def chat(request: ChatRequest):
                     system_prompt=request.system_prompt,
                     execution_id=request.execution_id,
                     isolated_session=request.isolated_session,
+                    **execute_kwargs,
                 )
             except HTTPException as exc:
                 agent_state.record_task_finish(success=False)
@@ -93,7 +129,7 @@ async def chat(request: ChatRequest):
                     request.execution_id,
                     result_callback._envelope_from_http_exception(exc),
                 )
-                raise
+                _reraise_with_auth_override(exc, auth_override)
             except BaseException:
                 agent_state.record_task_finish(success=False)
                 raise
@@ -136,6 +172,7 @@ async def chat(request: ChatRequest):
                 agent_state.session_context_window = metadata.context_window
 
             logger.info(f"[Chat] Execution lock releasing after completion")
+            _stamp_auth_override(response, auth_override)
 
             # Return enhanced response with execution log and session stats
             # Use raw_messages (full Claude Code JSON transcript) for execution log viewer compatibility
@@ -160,7 +197,7 @@ async def chat(request: ChatRequest):
 
 
 @router.post("/api/task")
-async def execute_task(request: ParallelTaskRequest):
+async def execute_task(request: ParallelTaskRequest, response: Response = None):
     """
     Execute a stateless task in parallel mode (no conversation context).
 
@@ -197,6 +234,10 @@ async def execute_task(request: ParallelTaskRequest):
 
     # Execute via runtime adapter in headless mode (no lock, no chat session)
     runtime = get_runtime()
+    # #3470: a SUB-003 re-issue's per-spawn credential (Claude only). The 202
+    # path above never carries one — the walk is inline-only (#1083 boundary).
+    auth_override = _accept_auth_override(request, runtime)
+    execute_kwargs = {"auth_override": auth_override} if auth_override is not None else {}
     # #2433: register as ACCEPTED before anything can queue. The headless
     # executor's `Popen` + `registry.register()` run inside a pool thread, so a
     # request beyond the pool size was counted in /health `active_tasks` but
@@ -224,6 +265,7 @@ async def execute_task(request: ParallelTaskRequest):
             resume_session_id=request.resume_session_id,  # Resume previous session (EXEC-023)
             persist_session=bool(request.persist_session),  # Session tab: write JSONL for future --resume
             images=request.images,  # Vision images from channel adapters (#562)
+            **execute_kwargs,
         )
     except HTTPException as exc:
         # #679 (F3): a terminated turn that surfaces a non-auth/non-rate terminal
@@ -251,6 +293,7 @@ async def execute_task(request: ParallelTaskRequest):
         )
         if is_cancel:
             logger.info(f"[Task] Task {request.execution_id} cancelled by user (status {exc.status_code})")
+            _stamp_auth_override(response, auth_override)
             return {
                 "response": exc.detail if isinstance(exc.detail, str) else "",
                 "execution_log": [],
@@ -259,7 +302,7 @@ async def execute_task(request: ParallelTaskRequest):
                 "status": "cancelled",
                 "timestamp": datetime.now().isoformat(),
             }
-        raise
+        _reraise_with_auth_override(exc, auth_override)
     except BaseException:
         agent_state.record_task_finish(success=False)
         raise
@@ -290,6 +333,7 @@ async def execute_task(request: ParallelTaskRequest):
         result_callback._cancelled_override(success_envelope) if cancelled else success_envelope,
     )
 
+    _stamp_auth_override(response, auth_override)
     # raw_messages contains the full Claude Code JSON stream (init, assistant, user, result)
     # This is the complete execution transcript showing thinking, tool calls, and results
     return {

docker/base-image/agent_server/routers/skills.py

@@ -17,6 +17,17 @@
 logger = logging.getLogger(__name__)
 router = APIRouter()
 
+# trinity-enterprise#754: the platform marks every library skill it delivers
+# with this file (backend `skill_packaging.META_FILENAME`); a skill directory
+# without it is the agent's own (#2914). Duplicated because this image cannot
+# import the backend — `tests/unit/test_ent754_agent_server_skillinfo.py`
+# keeps the two copies equal.
+PLATFORM_MARKER_FILENAME = ".trinity-skill.json"
+
+# trinity-enterprise#753: the values `approval:` may take (backend
+# `skill_packaging.APPROVAL_VALUES`; the same test keeps them equal).
+APPROVAL_VALUES = frozenset({"recommended"})
+
 
 class SkillInfo(BaseModel):
     """Information about a single skill/playbook."""
@@ -32,6 +43,13 @@ class SkillInfo(BaseModel):
     allowed_tools: Optional[List[str]] = None
     argument_hint: Optional[str] = None
     has_schedule: bool = False  # Placeholder for future schedule integration
+    # trinity-enterprise#754 — where the skill came from, and what its author
+    # recommends. Informational ONLY: the agent can write both the marker and
+    # its own frontmatter, so neither may decide access; the platform's gate
+    # map stays the authority.
+    source: Optional[str] = None    # "platform" (marker present) | "agent"
+    dir: Optional[str] = None       # the directory name; the gate fingerprint resolves it first
+    approval: Optional[str] = None  # "recommended" | None
 
 
 class SkillsResponse(BaseModel):
@@ -232,6 +250,31 @@ def _field(skill_md: Path, field: str, value: Any, normalize) -> Any:
         return None
 
 
+def _source(skill_dir: Path) -> str:
+    """trinity-enterprise#754: `platform` when the platform's marker is in the
+    directory (the same `isfile` test the backend's managed-skill listing and
+    the gate fingerprint use), else `agent`."""
+    return "platform" if (skill_dir / PLATFORM_MARKER_FILENAME).is_file() else "agent"
+
+
+def _approval(skill_md: Path, frontmatter: Dict[str, Any]) -> Optional[str]:
+    """trinity-enterprise#754: `approval:` read exactly as the backend's
+    `skill_packaging.extract_contract` reads it — a `trinity:` mapping first,
+    then the flat key, the first non-null value winning — and kept only when
+    it is in the closed set. Anything else (YAML 1.1 hands `yes` as a bool and
+    an unquoted date as a date) is skipped and warned once."""
+    block = frontmatter.get('trinity')
+    sources = ([block] if isinstance(block, dict) else []) + [frontmatter]
+    raw = next((s['approval'] for s in sources if s.get('approval') is not None), None)
+    if raw is None:
+        return None
+    if isinstance(raw, str) and raw.strip().lower() in APPROVAL_VALUES:
+        return raw.strip().lower()
+    _report_skipped_field(skill_md, 'approval', raw,
+                          'expected one of: ' + ', '.join(sorted(APPROVAL_VALUES)))
+    return None
+
+
 def scan_skills_directory(skills_dir: Path) -> List[SkillInfo]:
     """
     Scan a skills directory for subdirectories containing SKILL.md files.
@@ -295,7 +338,10 @@ def scan_skills_directory(skills_dir: Path) -> List[SkillInfo]:
                 automation=_field(skill_md, 'automation', frontmatter.get('automation'), _coerce_optional_str),
                 allowed_tools=_field(skill_md, 'allowed-tools', frontmatter.get('allowed-tools'), normalize_allowed_tools),
                 argument_hint=_field(skill_md, 'argument-hint', frontmatter.get('argument-hint'), _coerce_argument_hint),
-                has_schedule=False  # TODO: Check if schedule exists for this skill
+                has_schedule=False,  # TODO: Check if schedule exists for this skill
+                source=_source(entry),
+                dir=entry.name,
+                approval=_approval(skill_md, frontmatter),
             )
             skills.append(skill)
 
@@ -308,6 +354,10 @@ def scan_skills_directory(skills_dir: Path) -> List[SkillInfo]:
                 name=entry.name,
                 description=None,
                 path=_display_path(skill_md),
+                # #754: both come from the filesystem, so a card can still be
+                # placed in its section when the file itself is unreadable.
+                source=_source(entry),
+                dir=entry.name,
             ))
 
     return skills
@@ -326,11 +376,12 @@ async def list_skills():
     """
     home_dir = Path('/home/developer')
 
-    # Skills directories to scan
-    skill_paths = [
+    # Skills directories to scan. The server runs as `developer`, so in the
+    # agent container both are one folder: it is scanned and named once.
+    skill_paths = list(dict.fromkeys([
         home_dir / '.claude' / 'skills',
         Path.home() / '.claude' / 'skills'  # Personal skills
-    ]
+    ]))
 
     all_skills: List[SkillInfo] = []
     scanned_paths: List[str] = []