Complete WorkBuddy risk controls

app/admin_api.py

@@ -67,7 +67,7 @@ def _public_credential(item):
               "trial_supported", "trial",
               "daily_chat_supported", "auto_daily_chat", "daily_chat",
               "growth_supported", "auto_growth", "growth_auto",
-              "token_expired", "token_expires_at", "last_refresh_time", "sessions", "sticky_sessions", "last_error_code", "last_error_body", "auto_disabled", "usage_stats"}
+              "token_expired", "token_expires_at", "last_refresh_time", "sessions", "sticky_sessions", "last_error_code", "last_error_body", "auto_disabled", "restricted", "restricted_until", "usage_stats"}
     result = {key: value for key, value in item.items() if key in fields}
     identity = item.get("account_key") or item.get("id")
     if identity and ("/" in str(identity) or "\\" in str(identity)):

app/control_store.py

@@ -267,7 +267,10 @@ def set_credential(self, account_key, enabled):
         def change(state):
             record = state["credentials"].setdefault(account_key, {})
             record["enabled"] = enabled
-            record.pop("auto_disabled", None)  # Any manual toggle clears the provenance.
+            # Any manual toggle clears automatic restriction provenance. The reason/body/time
+            # fields are evidence for the disabled state and must not survive an appeal release.
+            for key in ("auto_disabled", "auto_reason", "auto_body", "auto_at"):
+                record.pop(key, None)
         return self._update(None, change)
 
     def disable_credential_auto(self, account_key, reason="", body=None, at=None):

app/desensitize.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import re
+from copy import deepcopy
 from typing import Any, Iterable
 
 from app.harness_context import parse_harness_text
@@ -142,7 +143,14 @@ def _template_pattern(text: str) -> re.Pattern:
 _DSH_MINIMAL_PATTERN = _template_pattern("You are a helpful software engineer assistant")
 _DSH_DIRECT_PATTERN = _template_pattern("You are DeepSeek Harness")
 _DSH_WEB_GUI_PATTERN = _template_pattern("DeepSeek Harness Web GUI")
+_DSH_WEB_GUI_FINGERPRINT_PATTERN = re.compile(
+    r"\bWeb GUI whose URL the runtime context names\.\s+When the user refers",
+    re.IGNORECASE,
+)
 _CHANNEL_IDENTITY_FIXED = _DSH_WORKBUDDY_IDENTITY
+_CHANNEL_SYSTEM_DEFAULT = "You are a helpful assistant."
+_CHANNEL_INSTRUCTIONS_OPEN = "<instructions>\n"
+_CHANNEL_INSTRUCTIONS_CLOSE = "\n</instructions>"
 
 WORKBUDDY_SYSTEM_PROMPT = (
     "You are WorkBuddy, a powerful AI assistant.\n\n"
@@ -266,7 +274,11 @@ def desensitize_channel_text(text: str) -> str:
     text = _CHANNEL_IDENTITY_PATTERN.sub(_CHANNEL_IDENTITY_FIXED, text)
     text = _DSH_MINIMAL_PATTERN.sub(_CHANNEL_IDENTITY_FIXED, text)
     text = _DSH_DIRECT_PATTERN.sub(_CHANNEL_IDENTITY_FIXED, text)
-    return _DSH_WEB_GUI_PATTERN.sub("WorkBuddy Web GUI", text)
+    text = _DSH_WEB_GUI_PATTERN.sub("WorkBuddy Web GUI", text)
+    return _DSH_WEB_GUI_FINGERPRINT_PATTERN.sub(
+        "Web GUI whose URL the runtime context names. The interface, when the user refers",
+        text,
+    )
 
 
 def _iter_text_blocks(content):
@@ -467,6 +479,75 @@ def desensitize_channel_body(body: dict) -> dict:
     return {**body, "messages": messages}
 
 
+def _instruction_blocks(contents: list) -> list[dict]:
+    """Wrap system content as user-visible instructions without dropping non-text blocks."""
+    blocks: list[dict] = [{"type": "text", "text": _CHANNEL_INSTRUCTIONS_OPEN}]
+    for index, content in enumerate(contents):
+        if index:
+            blocks.append({"type": "text", "text": "\n\n"})
+        if isinstance(content, str):
+            blocks.append({"type": "text", "text": content})
+        elif isinstance(content, list):
+            blocks.extend(deepcopy(content))
+        elif content is not None:
+            blocks.append({"type": "text", "text": str(content)})
+    blocks.append({"type": "text", "text": _CHANNEL_INSTRUCTIONS_CLOSE})
+    return blocks
+
+
+def _prepend_user_instructions(message: dict, contents: list) -> dict:
+    """Prepend wrapped instructions to one user message while preserving its shape."""
+    moved = _instruction_blocks(contents)
+    existing = message.get("content")
+    updated = dict(message)
+    if isinstance(existing, str):
+        if any(block.get("type") != "text" for block in moved):
+            updated["content"] = moved + ([{"type": "text", "text": existing}] if existing else [])
+        else:
+            text = "".join(block.get("text", "") for block in moved)
+            updated["content"] = text + ("\n\n" + existing if existing else "")
+    elif isinstance(existing, list):
+        updated["content"] = moved + deepcopy(existing)
+    elif existing is None:
+        updated["content"] = moved
+    else:
+        updated["content"] = moved + [{"type": "text", "text": str(existing)}]
+    return updated
+
+
+def system_as_user_body(body: dict) -> dict:
+    """Move a refused system prompt into the first user message for one bounded retry."""
+    messages = body.get("messages")
+    if not isinstance(messages, list) or not messages:
+        return body
+    leading: list = []
+    index = 0
+    while index < len(messages) and isinstance(messages[index], dict) \
+            and messages[index].get("role") == "system":
+        leading.append(messages[index].get("content"))
+        index += 1
+    rest = [deepcopy(message) for message in messages[index:]]
+    later_system = False
+    for message in rest:
+        if isinstance(message, dict) and message.get("role") == "system":
+            message["role"] = "user"
+            later_system = True
+    meaningful = any(_content_to_text(content).strip() for content in leading)
+    if meaningful and leading and not (
+        len(leading) == 1 and _content_to_text(leading[0]).strip() == _CHANNEL_SYSTEM_DEFAULT
+    ):
+        user_index = next((i for i, message in enumerate(rest)
+                           if isinstance(message, dict) and message.get("role") == "user"), None)
+        moved = _instruction_blocks(leading)
+        if user_index is None:
+            rest.insert(0, {"role": "user", "content": moved})
+        else:
+            rest[user_index] = _prepend_user_instructions(rest[user_index], leading)
+    if not meaningful and not later_system:
+        return body
+    return {**body, "messages": [{"role": "system", "content": _CHANNEL_SYSTEM_DEFAULT}, *rest]}
+
+
 def apply_system_prompt_mode(body: dict, profile: str, mode: str) -> dict:
     """Apply an optional profile prompt without mutating canonical messages."""
     if mode == "empty":

app/gateway_management.py

@@ -47,17 +47,21 @@ def admin_credential_inventory(self):
                 record = policy.credential(entry)
                 enabled = record.get("enabled", True)
                 until = entry.get("fail_until", 0)
+                restricted = entry.get("restricted") is True
+                restricted_until = entry.get("restricted_until") if restricted else None
                 cooldowns = [{"model": model, "until": deadline,
                               "remaining_seconds": max(0, round(deadline - now))}
                              for (cid, model), deadline in pool._model_fail.items()
                              if cid == path and deadline > now]
                 # Provenance lives in the store so it survives restarts and cooldown resets.
-                auto_disabled = record.get("auto_disabled") is True and not enabled
-                state = ("auto_disabled" if auto_disabled else
+                auto_disabled = record.get("auto_disabled") is True and not enabled and not restricted
+                state = ("restricted" if restricted else
+                         "auto_disabled" if auto_disabled else
                          "disabled" if not enabled else "error" if row.get("error") else
                          "circuit_open" if until > now else "expired" if row.get("token_expired") else "ready")
                 row.update(id=identity, name=Path(path).name, enabled=enabled, health=state,
-                           auto_disabled=auto_disabled,
+                           auto_disabled=auto_disabled, restricted=restricted,
+                           restricted_until=restricted_until,
                            auto_checkin=model_policy.credential_auto_checkin(self.CONFIG, entry),
                            checkin=checkin.view(balance.get("checkin") or {}),
                            auto_travel=model_policy.credential_auto_travel(self.CONFIG, entry),
@@ -76,7 +80,8 @@ def admin_credential_inventory(self):
                                              if daily_chats else {}) or {}).get("growth_auto") or {},
                            fail_until=until, cooldown_until=until,
                            cooldown_remaining=max(0, round(until - now)),
-                           last_error_code=("http_401" if entry.get("last_error") == "backend HTTP 401" else
+                           last_error_code=("restricted" if restricted else
+                                            "http_401" if entry.get("last_error") == "backend HTTP 401" else
                                             "http_403" if entry.get("last_error") == "backend HTTP 403" else
                                             "credential_error" if entry.get("last_error") else None),
                            last_failure_at=entry.get("last_failure_at"),