fix(api): secure jobs listing contract

api/routes.py

@@ -66,6 +66,10 @@ except ModuleNotFoundError:  # pragma: no cover (optional for unit tests)
 
 PACK_NAME = PACK_VERSION = PACK_START_TIME = LOG_FILE = get_api_key = None  # type: ignore
 metrics = tail_log = require_observability_access = check_rate_limit = trace_store = None  # type: ignore
+require_admin_token = resolve_token_info = emit_audit_event = None  # type: ignore
+jobs_request_tenant_scope = normalize_jobs_query = build_jobs_audit_details = None  # type: ignore
+SAFE_JOB_AUDIT_OUTCOMES = None  # type: ignore
+JobsSecurityError = TenantBoundaryError = None  # type: ignore
 get_executor_diagnostics = None  # type: ignore
 webhook_handler = webhook_submit_handler = webhook_validate_handler = capabilities_handler = preflight_handler = None  # type: ignore
 pnginfo_handler = None  # type: ignore  # R168
@@ -276,11 +280,43 @@ if web is not None:
     # CRITICAL: These imports MUST remain present.
     # If edited out, module-level placeholders stay as None and handlers raise at runtime
     # (e.g., TypeError: 'NoneType' object is not callable), producing noisy aiohttp tracebacks.
-    (require_admin_token, require_observability_access) = import_attrs_dual(
+    (require_admin_token, require_observability_access, resolve_token_info) = (
+        import_attrs_dual(
+            __package__,
+            "..services.access_control",
+            "services.access_control",
+            (
+                "require_admin_token",
+                "require_observability_access",
+                "resolve_token_info",
+            ),
+        )
+    )
+    (emit_audit_event,) = import_attrs_dual(
         __package__,
-        "..services.access_control",
-        "services.access_control",
-        ("require_admin_token", "require_observability_access"),
+        "..services.audit",
+        "services.audit",
+        ("emit_audit_event",),
+    )
+    (
+        JobsSecurityError,
+        SAFE_JOB_AUDIT_OUTCOMES,
+        TenantBoundaryError,
+        build_jobs_audit_details,
+        jobs_request_tenant_scope,
+        normalize_jobs_query,
+    ) = import_attrs_dual(
+        __package__,
+        "..services.jobs_security",
+        "services.jobs_security",
+        (
+            "JobsSecurityError",
+            "SAFE_JOB_AUDIT_OUTCOMES",
+            "TenantBoundaryError",
+            "build_jobs_audit_details",
+            "jobs_request_tenant_scope",
+            "normalize_jobs_query",
+        ),
     )
     (tail_log,) = import_attrs_dual(
         __package__,
@@ -630,17 +666,82 @@ async def logs_tail_handler(request: web.Request) -> web.Response:
     auth=AuthTier.ADMIN,
     risk=RiskTier.LOW,
     summary="List jobs",
-    description="Stub endpoint for job listing.",
+    description="Admin-authorized jobs list contract (compatibility stub until adapter availability).",
     audit="jobs.list",
     plane=RoutePlane.ADMIN,
 )
 async def jobs_handler(request: web.Request) -> web.Response:
     """
-    GET /moltbot/jobs
-    Stub endpoint for job listing (not implemented yet).
+    GET /openclaw/jobs (legacy: /moltbot/jobs).
+    This handler secures the compatibility stub before a read adapter is wired.
     """
     if web is None:
         raise RuntimeError("aiohttp not available")
+
+    token_info = resolve_token_info(request)
+
+    if not check_rate_limit(request, "admin"):
+        _emit_jobs_list_audit(
+            request=request,
+            token_info=token_info,
+            outcome="rate_limit",
+            status_code=429,
+            reason="jobs_rate_limited",
+        )
+        return build_rate_limit_response(
+            request,
+            "admin",
+            web_module=web,
+            error="jobs_rate_limited",
+            include_ok=True,
+        )
+
+    # CRITICAL: endpoint metadata is descriptive; this explicit guard is the
+    # runtime boundary that must remain before any queue/history access.
+    allowed, _error = require_admin_token(request)
+    if not allowed:
+        _emit_jobs_list_audit(
+            request=request,
+            token_info=token_info,
+            outcome="deny",
+            status_code=403,
+            reason="jobs_admin_required",
+        )
+        return web.json_response(
+            {"ok": False, "error": "jobs_admin_required"}, status=403
+        )
+
+    try:
+        with jobs_request_tenant_scope(request, token_info):
+            normalize_jobs_query(request.query)
+            _emit_jobs_list_audit(
+                request=request,
+                token_info=token_info,
+                outcome="allow",
+                status_code=200,
+                reason="stub",
+            )
+    except TenantBoundaryError as exc:
+        _emit_jobs_list_audit(
+            request=request,
+            token_info=token_info,
+            outcome="deny",
+            status_code=403,
+            reason=exc.code,
+        )
+        return web.json_response({"ok": False, "error": exc.code}, status=403)
+    except JobsSecurityError:
+        _emit_jobs_list_audit(
+            request=request,
+            token_info=token_info,
+            outcome="error",
+            status_code=400,
+            reason="jobs_query_invalid",
+        )
+        return web.json_response(
+            {"ok": False, "error": "jobs_query_invalid"}, status=400
+        )
+
     return web.json_response(
         {
             "ok": True,
@@ -651,6 +752,29 @@ async def jobs_handler(request: web.Request) -> web.Response:
     )
 
 
+def _emit_jobs_list_audit(
+    *,
+    request,
+    token_info,
+    outcome: str,
+    status_code: int,
+    reason: str,
+    **counts,
+) -> None:
+    """Emit only content-free jobs audit dimensions."""
+
+    safe_outcome = outcome if outcome in SAFE_JOB_AUDIT_OUTCOMES else "error"
+    emit_audit_event(
+        action="jobs.list",
+        target="jobs",
+        outcome=safe_outcome,
+        token_info=token_info,
+        status_code=status_code,
+        details=build_jobs_audit_details(reason, **counts),
+        request=request,
+    )
+
+
 @endpoint_metadata(
     auth=AuthTier.ADMIN,
     risk=RiskTier.MEDIUM,

docs/openapi.yaml

@@ -1,7 +1,7 @@
 openapi: "3.0.3"
 info:
   title: "ComfyUI-OpenClaw API"
-  version: "1.0.13"
+  version: "1.0.14"
   description: "Generated from docs/release/api_contract.md."
 servers:
   - url: "/openclaw"
@@ -145,16 +145,16 @@ paths:
   /jobs:
     get:
       operationId: "get_jobs"
-      summary: "List recent jobs (Stub/Not Implemented)."
+      summary: "List recent jobs (Admin-authorized compatibility stub until the bounded read adapter is available)."
       responses:
         200:
           description: "OK"
-      x-openclaw-auth: "Observability"
+      x-openclaw-auth: "Admin"
       x-openclaw-section: "1.1 Core Observability & System"
       x-openclaw-legacy-path: "/moltbot/jobs"
-      x-openclaw-auth-tier: "observability"
+      x-openclaw-auth-tier: "admin"
       security:
-        - OpenClawObservabilityToken:
+        - OpenClawAdminToken:
             []
   /preflight:
     post:

docs/release/api_contract.md

@@ -1,8 +1,8 @@
 # OpenClaw API Contract (v1)
 
 > **Status**: normative
-> **Version**: 1.0.13
-> **Date**: 2026-07-08
+> **Version**: 1.0.14
+> **Date**: 2026-07-10
 
 This document defines the public API contract for OpenClaw. It serves as the authoritative baseline for client compatibility and breaking change policies.
 
@@ -45,7 +45,7 @@ All new integrations should use the `/openclaw/` prefix. Use of `/moltbot/` is d
 | `GET` | `/events/stream` | `/moltbot/events/stream` | Observability | SSE stream of job lifecycle events with resume support. |
 | `GET` | `/config` | `/moltbot/config` | Observability | Read-only view of sanitized provider config. |
 | `PUT` | `/config` | `/moltbot/config` | Admin | Update system configuration. |
-| `GET` | `/jobs` | `/moltbot/jobs` | Observability | List recent jobs (Stub/Not Implemented). |
+| `GET` | `/jobs` | `/moltbot/jobs` | Admin | List recent jobs (Admin-authorized compatibility stub until the bounded read adapter is available). |
 | `POST` | `/preflight` | `/moltbot/preflight` | Admin | Analyze a workflow or API prompt payload for missing nodes/models and portability diagnostics. |
 | `GET` | `/preflight/inventory` | `/moltbot/preflight/inventory` | Admin | Snapshot-first inventory of nodes/models for operator diagnostics, including refresh-state metadata. |
 

services/jobs_security.py

@@ -0,0 +1,389 @@
+"""Security boundary for jobs ownership filtering and list projections."""
+
+from __future__ import annotations
+
+import contextlib
+import math
+from collections.abc import Iterable, Mapping
+from dataclasses import dataclass
+from typing import Any, Iterator, Literal
+
+from .management_query import normalize_limit_offset
+from .tenant_context import (
+    DEFAULT_TENANT_ID,
+    TenantBoundaryError,
+    extract_tenant_from_headers,
+    is_multi_tenant_enabled,
+    normalize_tenant_id,
+    request_tenant_scope,
+)
+
+JobSource = Literal["queue", "history"]
+
+ALLOWED_JOB_STATUSES = frozenset(
+    {"pending", "in_progress", "completed", "failed", "cancelled"}
+)
+SAFE_JOB_SUMMARY_FIELDS = frozenset(
+    {
+        "id",
+        "status",
+        "priority",
+        "create_time",
+        "execution_start_time",
+        "execution_end_time",
+        "outputs_count",
+        "workflow_id",
+    }
+)
+MAX_JOB_IDENTIFIER_LENGTH = 128
+MAX_OUTPUT_COUNT = 1_000_000
+MAX_ABSOLUTE_NUMBER = 1_000_000_000_000_000_000
+DEFAULT_JOBS_LIMIT = 50
+MAX_JOBS_LIMIT = 200
+MAX_JOBS_OFFSET = 10_000
+MAX_JOBS_SOURCE_WINDOW = 10_000
+ALLOWED_JOB_SORT_FIELDS = frozenset({"created_at", "execution_duration"})
+ALLOWED_JOB_SORT_ORDERS = frozenset({"asc", "desc"})
+ALLOWED_JOB_QUERY_FIELDS = frozenset(
+    {"status", "workflow_id", "sort_by", "sort_order", "limit", "offset"}
+)
+SAFE_JOB_AUDIT_OUTCOMES = frozenset(
+    {"allow", "deny", "rate_limit", "unsupported", "error"}
+)
+SAFE_JOB_AUDIT_REASONS = frozenset(
+    {
+        "stub",
+        "jobs_listed",
+        "jobs_admin_required",
+        "jobs_rate_limited",
+        "jobs_query_invalid",
+        "jobs_host_contract_unsupported",
+        "jobs_backend_unavailable",
+        "tenant_required",
+        "tenant_mismatch",
+        "tenant_invalid",
+        "jobs_error",
+    }
+)
+SAFE_JOB_AUDIT_COUNT_FIELDS = frozenset(
+    {"returned_count", "excluded_count", "malformed_count"}
+)
+_BOOTSTRAP_TOKEN_IDS = frozenset({"env-admin", "local-admin", "local-internal"})
+
+
+class JobsSecurityError(ValueError):
+    """Raised when an upstream jobs value cannot cross the list boundary safely."""
+
+    def __init__(self, code: str, message: str):
+        super().__init__(message)
+        self.code = code
+
+
+@dataclass(frozen=True)
+class VisibilityFilterResult:
+    """Immutable result of filtering raw queue/history records by ownership."""
+
+    records: tuple[Any, ...]
+    excluded_count: int
+    malformed_count: int
+
+
+@dataclass(frozen=True)
+class JobsQueryWarning:
+    """Bounded pagination warning without echoing raw request input."""
+
+    code: str
+    field: str
+    normalized: int
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "code": self.code,
+            "field": self.field,
+            "normalized": self.normalized,
+        }
+
+
+@dataclass(frozen=True)
+class JobsQuery:
+    """Normalized immutable jobs list query."""
+
+    status: str | None
+    workflow_id: str | None
+    sort_by: str
+    sort_order: str
+    limit: int
+    offset: int
+    warnings: tuple[JobsQueryWarning, ...]
+
+    def to_pagination(self) -> dict[str, Any]:
+        return {
+            "limit": self.limit,
+            "offset": self.offset,
+            "warnings": [warning.to_dict() for warning in self.warnings],
+        }
+
+
+@contextlib.contextmanager
+def jobs_request_tenant_scope(request: Any, token_info: Any) -> Iterator[Any]:
+    """Bind an explicit jobs tenant context without bootstrap default fallback."""
+
+    if is_multi_tenant_enabled():
+        headers = getattr(request, "headers", None)
+        header_tenant = (
+            extract_tenant_from_headers(headers)
+            if isinstance(headers, Mapping)
+            else None
+        )
+        token_id = str(getattr(token_info, "token_id", "") or "")
+        token_tenant = str(
+            getattr(token_info, "tenant_id", DEFAULT_TENANT_ID) or DEFAULT_TENANT_ID
+        )
+
+        # CRITICAL: bootstrap/local token resolution uses `default` when the
+        # tenant header is absent. Jobs must not reinterpret that fallback as
+        # an explicit cross-job tenant authorization.
+        if (
+            header_tenant is None
+            and token_id in _BOOTSTRAP_TOKEN_IDS
+            and token_tenant == DEFAULT_TENANT_ID
+        ):
+            raise TenantBoundaryError(
+                "tenant_required",
+                "Explicit tenant context is required for jobs in multi-tenant mode.",
+            )
+
+    with request_tenant_scope(
+        request=request,
+        token_info=token_info,
+        allow_default_when_missing=False,
+    ) as context:
+        yield context
+
+
+def filter_visible_job_records(
+    records: Iterable[Any],
+    *,
+    source: JobSource,
+    tenant_id: str,
+    multi_tenant: bool,
+) -> VisibilityFilterResult:
+    """Filter raw records before upstream normalization or pagination."""
+
+    if source not in {"queue", "history"}:
+        raise JobsSecurityError("jobs_source_invalid", "Unsupported jobs source.")
+
+    materialized = tuple(records)
+    if not multi_tenant:
+        return VisibilityFilterResult(
+            records=materialized,
+            excluded_count=0,
+            malformed_count=0,
+        )
+
+    expected_tenant = normalize_tenant_id(tenant_id)
+    visible: list[Any] = []
+    malformed_count = 0
+
+    for record in materialized:
+        owner, malformed = _extract_owner(record, source=source)
+        if malformed:
+            malformed_count += 1
+        if owner == expected_tenant:
+            visible.append(record)
+
+    return VisibilityFilterResult(
+        records=tuple(visible),
+        excluded_count=len(materialized) - len(visible),
+        malformed_count=malformed_count,
+    )
+
+
+def normalize_jobs_query(query: Mapping[str, Any]) -> JobsQuery:
+    """Normalize the frozen jobs filter/sort/pagination contract."""
+
+    if not isinstance(query, Mapping):
+        raise JobsSecurityError("jobs_query_invalid", "Jobs query must be a mapping.")
+
+    unknown = set(query) - ALLOWED_JOB_QUERY_FIELDS
+    if unknown:
+        raise JobsSecurityError(
+            "jobs_query_invalid", "Jobs query contains unsupported fields."
+        )
+
+    status = _optional_query_value(query.get("status"), field="status")
+    if status is not None and status not in ALLOWED_JOB_STATUSES:
+        raise JobsSecurityError("jobs_query_invalid", "Unsupported jobs status.")
+
+    workflow_id = _optional_query_value(query.get("workflow_id"), field="workflow_id")
+    sort_by = (
+        _optional_query_value(query.get("sort_by"), field="sort_by") or "created_at"
+    )
+    if sort_by not in ALLOWED_JOB_SORT_FIELDS:
+        raise JobsSecurityError("jobs_query_invalid", "Unsupported jobs sort field.")
+    sort_order = (
+        _optional_query_value(query.get("sort_order"), field="sort_order") or "desc"
+    )
+    if sort_order not in ALLOWED_JOB_SORT_ORDERS:
+        raise JobsSecurityError("jobs_query_invalid", "Unsupported jobs sort order.")
+
+    page = normalize_limit_offset(
+        dict(query),
+        default_limit=DEFAULT_JOBS_LIMIT,
+        max_limit=MAX_JOBS_LIMIT,
+        default_offset=0,
+        max_offset=MAX_JOBS_OFFSET,
+    )
+    warnings = tuple(
+        JobsQueryWarning(
+            code=str(warning.get("code") or "jobs_query_normalized"),
+            field=str(warning.get("field") or "query"),
+            normalized=int(warning.get("normalized") or 0),
+        )
+        for warning in page.warnings
+    )
+    return JobsQuery(
+        status=status,
+        workflow_id=workflow_id,
+        sort_by=sort_by,
+        sort_order=sort_order,
+        limit=page.limit,
+        offset=page.offset,
+        warnings=warnings,
+    )
+
+
+def build_jobs_audit_details(reason: Any, **counts: Any) -> dict[str, Any]:
+    """Build content-free jobs audit details from safe codes and aggregate counts."""
+
+    safe_reason = str(reason or "")
+    if safe_reason not in SAFE_JOB_AUDIT_REASONS:
+        safe_reason = "jobs_error"
+    details: dict[str, Any] = {"reason": safe_reason}
+    for field in SAFE_JOB_AUDIT_COUNT_FIELDS:
+        value = counts.get(field)
+        if isinstance(value, bool) or not isinstance(value, int):
+            continue
+        details[field] = max(0, min(value, MAX_JOBS_SOURCE_WINDOW))
+    return details
+
+
+def project_job_summary(job: Mapping[str, Any]) -> dict[str, Any]:
+    """Project an upstream normalized job onto the frozen list allowlist."""
+
+    if not isinstance(job, Mapping):
+        raise JobsSecurityError("jobs_record_invalid", "Job must be an object.")
+
+    projected: dict[str, Any] = {
+        "id": _bounded_identifier(job.get("id"), field="id"),
+        "status": _validated_status(job.get("status")),
+    }
+
+    for field in (
+        "priority",
+        "create_time",
+        "execution_start_time",
+        "execution_end_time",
+    ):
+        if field in job and job[field] is not None:
+            projected[field] = _bounded_number(job[field], field=field)
+
+    if "outputs_count" in job and job["outputs_count"] is not None:
+        outputs_count = job["outputs_count"]
+        if (
+            isinstance(outputs_count, bool)
+            or not isinstance(outputs_count, int)
+            or outputs_count < 0
+            or outputs_count > MAX_OUTPUT_COUNT
+        ):
+            raise JobsSecurityError(
+                "jobs_record_invalid", "outputs_count is outside the safe bound."
+            )
+        projected["outputs_count"] = outputs_count
+
+    if "workflow_id" in job and job["workflow_id"] is not None:
+        projected["workflow_id"] = _bounded_identifier(
+            job["workflow_id"], field="workflow_id"
+        )
+
+    return projected
+
+
+def _extract_owner(record: Any, *, source: JobSource) -> tuple[str | None, bool]:
+    extra_data: Any
+    if source == "queue":
+        if not isinstance(record, (list, tuple)) or len(record) < 4:
+            return None, True
+        extra_data = record[3]
+    else:
+        if not isinstance(record, Mapping):
+            return None, True
+        prompt = record.get("prompt")
+        if not isinstance(prompt, (list, tuple)) or len(prompt) < 4:
+            return None, True
+        extra_data = prompt[3]
+
+    if not isinstance(extra_data, Mapping):
+        return None, True
+    openclaw = extra_data.get("openclaw")
+    if openclaw is None:
+        return None, False
+    if not isinstance(openclaw, Mapping):
+        return None, True
+    owner = openclaw.get("tenant_id")
+    if owner is None:
+        return None, False
+    try:
+        return normalize_tenant_id(owner), False
+    except TenantBoundaryError:
+        return None, True
+
+
+def _bounded_identifier(value: Any, *, field: str) -> str:
+    if not isinstance(value, str) or not value or value != value.strip():
+        raise JobsSecurityError(
+            "jobs_record_invalid", f"{field} must be a non-empty string."
+        )
+    if len(value) > MAX_JOB_IDENTIFIER_LENGTH:
+        raise JobsSecurityError(
+            "jobs_record_invalid", f"{field} exceeds the safe length bound."
+        )
+    return value
+
+
+def _optional_query_value(value: Any, *, field: str) -> str | None:
+    if value is None:
+        return None
+    if not isinstance(value, str) or not value or value != value.strip():
+        raise JobsSecurityError(
+            "jobs_query_invalid", f"{field} must be a non-empty string."
+        )
+    if len(value) > MAX_JOB_IDENTIFIER_LENGTH:
+        raise JobsSecurityError(
+            "jobs_query_invalid", f"{field} exceeds the safe length bound."
+        )
+    return value
+
+
+def _validated_status(value: Any) -> str:
+    if not isinstance(value, str) or value not in ALLOWED_JOB_STATUSES:
+        raise JobsSecurityError(
+            "jobs_record_invalid", "status is outside the jobs lifecycle contract."
+        )
+    return str(value)
+
+
+def _bounded_number(value: Any, *, field: str) -> int | float:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise JobsSecurityError("jobs_record_invalid", f"{field} must be numeric.")
+    # Check integer magnitude before float conversion; hostile large integers
+    # can otherwise raise OverflowError inside math.isfinite().
+    if isinstance(value, int):
+        valid = abs(value) <= MAX_ABSOLUTE_NUMBER
+    else:
+        valid = math.isfinite(value) and abs(value) <= MAX_ABSOLUTE_NUMBER
+    if not valid:
+        raise JobsSecurityError(
+            "jobs_record_invalid", f"{field} is outside the safe numeric bound."
+        )
+    return value

tests/skip_policy.json

@@ -18,6 +18,7 @@
     "tests.test_f57_slack_socket_mode_startup",
     "tests.test_connector_installation_registry",
     "tests.test_connector_callback_contract",
+    "tests.test_s100_jobs_security_contract",
     "tests.test_s70_ssrf_pinning_regression"
   ],
   "no_skip_module_metadata": {
@@ -89,6 +90,10 @@
       "reason": "Connector callback contract protects external ingress normalization and approval wiring.",
       "review_after": "2026-10-31"
     },
+    "tests.test_s100_jobs_security_contract": {
+      "reason": "Jobs authorization, tenant isolation, and privacy projection are security-critical and must remain non-skippable.",
+      "review_after": "2026-10-31"
+    },
     "tests.test_s70_ssrf_pinning_regression": {
       "reason": "SSRF pinning regression is a security-critical boundary test and must never degrade to skip coverage.",
       "review_after": "2026-10-31"

tests/test_s100_jobs_security_contract.py

@@ -0,0 +1,690 @@
+"""S100 jobs endpoint authorization, tenant, privacy, and audit contract tests."""
+
+from __future__ import annotations
+
+import importlib
+import json
+import os
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+try:
+    from aiohttp import web
+except ImportError:  # pragma: no cover
+    web = None
+
+from services.endpoint_manifest import AuthTier, get_metadata
+from tests.security_contract_assertions import assert_security_reject_contract
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def _request(
+    *,
+    remote: str = "203.0.113.10",
+    headers: dict[str, str] | None = None,
+    path: str = "/openclaw/jobs",
+    query: dict[str, str] | None = None,
+):
+    request = MagicMock()
+    request.remote = remote
+    request.headers = headers or {}
+    request.query = query or {}
+    request.path = path
+    return request
+
+
+def _decode(response) -> dict:
+    return json.loads(response.text)
+
+
+def _load_security_module(testcase: unittest.TestCase):
+    try:
+        return importlib.import_module("services.jobs_security")
+    except ModuleNotFoundError:
+        testcase.fail("services.jobs_security must define the S100 policy boundary")
+
+
+@unittest.skipIf(web is None, "aiohttp not installed")
+class TestJobsHandlerSecurity(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.env = patch.dict(os.environ, {}, clear=True)
+        self.env.start()
+        self.addCleanup(self.env.stop)
+
+    async def test_remote_without_token_is_denied_with_triple_assert(self):
+        from api import routes
+
+        request = _request()
+        with (
+            patch.object(routes, "check_rate_limit", return_value=True),
+            patch.object(routes, "emit_audit_event", create=True) as audit,
+        ):
+            response = await routes.jobs_handler(request)
+
+        assert_security_reject_contract(
+            self,
+            response=response,
+            expected_status=403,
+            expected_code="jobs_admin_required",
+            audit_mock=audit,
+            expected_action="jobs.list",
+            expected_outcome="deny",
+            expected_audit_status=403,
+            expected_reason="jobs_admin_required",
+        )
+
+    async def test_configured_admin_token_matrix(self):
+        from api import routes
+
+        os.environ["OPENCLAW_ADMIN_TOKEN"] = "configured-admin-value"
+        with (
+            patch.object(routes, "check_rate_limit", return_value=True),
+            patch.object(routes, "emit_audit_event", create=True) as audit,
+        ):
+            allowed = await routes.jobs_handler(
+                _request(headers={"X-OpenClaw-Admin-Token": "configured-admin-value"})
+            )
+            denied = await routes.jobs_handler(
+                _request(headers={"X-OpenClaw-Admin-Token": "wrong-value"})
+            )
+
+        self.assertEqual(allowed.status, 200)
+        self.assertTrue(_decode(allowed)["not_implemented"])
+        self.assertEqual(denied.status, 403)
+        self.assertEqual(_decode(denied)["error"], "jobs_admin_required")
+        outcomes = [call.kwargs.get("outcome") for call in audit.call_args_list]
+        self.assertIn("allow", outcomes)
+        self.assertIn("deny", outcomes)
+
+    async def test_loopback_same_origin_and_cross_origin_match_admin_policy(self):
+        from api import routes
+
+        with (
+            patch.object(routes, "check_rate_limit", return_value=True),
+            patch.object(routes, "emit_audit_event", create=True),
+        ):
+            same_origin = await routes.jobs_handler(
+                _request(
+                    remote="127.0.0.1",
+                    headers={"Sec-Fetch-Site": "same-origin"},
+                )
+            )
+            cross_origin = await routes.jobs_handler(
+                _request(
+                    remote="127.0.0.1",
+                    headers={"Sec-Fetch-Site": "cross-site"},
+                )
+            )
+
+        self.assertEqual(same_origin.status, 200)
+        self.assertEqual(cross_origin.status, 403)
+        self.assertEqual(_decode(cross_origin)["error"], "jobs_admin_required")
+
+    async def test_multi_tenant_requires_explicit_context(self):
+        from api import routes
+
+        os.environ.update(
+            {
+                "OPENCLAW_ADMIN_TOKEN": "configured-admin-value",
+                "OPENCLAW_MULTI_TENANT_ENABLED": "1",
+            }
+        )
+        with (
+            patch.object(routes, "check_rate_limit", return_value=True),
+            patch.object(routes, "emit_audit_event", create=True) as audit,
+        ):
+            response = await routes.jobs_handler(
+                _request(headers={"X-OpenClaw-Admin-Token": "configured-admin-value"})
+            )
+
+        assert_security_reject_contract(
+            self,
+            response=response,
+            expected_status=403,
+            expected_code="tenant_required",
+            audit_mock=audit,
+            expected_action="jobs.list",
+            expected_outcome="deny",
+            expected_audit_status=403,
+            expected_reason="tenant_required",
+        )
+
+    async def test_multi_tenant_exact_header_context_is_allowed(self):
+        from api import routes
+
+        os.environ.update(
+            {
+                "OPENCLAW_ADMIN_TOKEN": "configured-admin-value",
+                "OPENCLAW_MULTI_TENANT_ENABLED": "1",
+            }
+        )
+        request = _request(
+            headers={
+                "X-OpenClaw-Admin-Token": "configured-admin-value",
+                "X-OpenClaw-Tenant-Id": "team-a",
+            }
+        )
+        with (
+            patch.object(routes, "check_rate_limit", return_value=True),
+            patch.object(routes, "emit_audit_event", create=True) as audit,
+        ):
+            response = await routes.jobs_handler(request)
+
+        self.assertEqual(response.status, 200)
+        allow = [
+            call.kwargs
+            for call in audit.call_args_list
+            if call.kwargs.get("outcome") == "allow"
+        ]
+        self.assertEqual(len(allow), 1)
+        self.assertEqual(allow[0]["details"], {"reason": "stub"})
+
+    async def test_multi_tenant_invalid_and_mismatched_contexts_are_audited(self):
+        from api import routes
+
+        os.environ.update(
+            {
+                "OPENCLAW_ADMIN_TOKEN": "configured-admin-value",
+                "OPENCLAW_MULTI_TENANT_ENABLED": "1",
+            }
+        )
+        cases = (
+            (
+                _request(
+                    headers={
+                        "X-OpenClaw-Admin-Token": "configured-admin-value",
+                        "X-OpenClaw-Tenant-Id": "invalid tenant",
+                    }
+                ),
+                None,
+                "tenant_invalid",
+            ),
+            (
+                _request(headers={"X-OpenClaw-Tenant-Id": "team-b"}),
+                SimpleNamespace(token_id="kid-test", tenant_id="team-a"),
+                "tenant_mismatch",
+            ),
+        )
+        for request, forced_token, expected_code in cases:
+            with self.subTest(expected_code=expected_code):
+                with (
+                    patch.object(routes, "check_rate_limit", return_value=True),
+                    patch.object(routes, "emit_audit_event", create=True) as audit,
+                    patch.object(
+                        routes,
+                        "resolve_token_info",
+                        wraps=routes.resolve_token_info,
+                    ) as resolve,
+                    patch.object(
+                        routes,
+                        "require_admin_token",
+                        wraps=routes.require_admin_token,
+                    ) as require,
+                ):
+                    if forced_token is not None:
+                        resolve.return_value = forced_token
+                        require.return_value = (True, None)
+                    response = await routes.jobs_handler(request)
+
+                assert_security_reject_contract(
+                    self,
+                    response=response,
+                    expected_status=403,
+                    expected_code=expected_code,
+                    audit_mock=audit,
+                    expected_action="jobs.list",
+                    expected_outcome="deny",
+                    expected_audit_status=403,
+                    expected_reason=expected_code,
+                )
+
+    async def test_rate_limit_returns_standard_contract_and_safe_audit(self):
+        from api import routes
+
+        request = _request(
+            remote="127.0.0.1", headers={"Sec-Fetch-Site": "same-origin"}
+        )
+        with (
+            patch.object(routes, "check_rate_limit", return_value=False),
+            patch.object(routes, "build_rate_limit_response") as rate_response,
+            patch.object(routes, "emit_audit_event", create=True) as audit,
+        ):
+            rate_response.return_value = web.json_response(
+                {"ok": False, "error": "jobs_rate_limited"}, status=429
+            )
+            response = await routes.jobs_handler(request)
+
+        self.assertEqual(response.status, 429)
+        self.assertEqual(_decode(response)["error"], "jobs_rate_limited")
+        rate_response.assert_called_once()
+        event = audit.call_args.kwargs
+        self.assertEqual(event["action"], "jobs.list")
+        self.assertEqual(event["outcome"], "rate_limit")
+        self.assertEqual(event["details"], {"reason": "jobs_rate_limited"})
+
+    async def test_invalid_query_returns_triple_assert_error(self):
+        from api import routes
+
+        request = _request(
+            remote="127.0.0.1",
+            headers={"Sec-Fetch-Site": "same-origin"},
+            query={"status": "mystery"},
+        )
+        with (
+            patch.object(routes, "check_rate_limit", return_value=True),
+            patch.object(routes, "emit_audit_event", create=True) as audit,
+        ):
+            response = await routes.jobs_handler(request)
+
+        assert_security_reject_contract(
+            self,
+            response=response,
+            expected_status=400,
+            expected_code="jobs_query_invalid",
+            audit_mock=audit,
+            expected_action="jobs.list",
+            expected_outcome="error",
+            expected_audit_status=400,
+            expected_reason="jobs_query_invalid",
+        )
+
+    async def test_all_primary_browser_and_legacy_aliases_share_contract(self):
+        from api import routes
+
+        os.environ["OPENCLAW_ADMIN_TOKEN"] = "configured-admin-value"
+        server = MagicMock()
+        server.routes.get = MagicMock()
+        server.routes.post = MagicMock()
+        server.routes.put = MagicMock()
+        server.routes.delete = MagicMock()
+        server.app.router.add_route = MagicMock()
+
+        with (
+            patch.object(routes, "check_rate_limit", return_value=True),
+            patch.object(routes, "emit_audit_event", create=True),
+        ):
+            routes.register_routes(server)
+            registered = {
+                call.args[1]: call.args[2]
+                for call in server.app.router.add_route.call_args_list
+                if call.args[0] == "GET" and call.args[1].endswith("/jobs")
+            }
+            bodies = []
+            for path in (
+                "/openclaw/jobs",
+                "/api/openclaw/jobs",
+                "/moltbot/jobs",
+                "/api/moltbot/jobs",
+            ):
+                response = await registered[path](
+                    _request(
+                        path=path,
+                        headers={"X-OpenClaw-Admin-Token": "configured-admin-value"},
+                    )
+                )
+                self.assertEqual(response.status, 200)
+                bodies.append(_decode(response))
+                if "moltbot" in path:
+                    self.assertEqual(response.headers["Deprecation"], "true")
+                    self.assertEqual(
+                        response.headers["X-OpenClaw-Canonical-Path"],
+                        path.replace("moltbot", "openclaw"),
+                    )
+
+        self.assertTrue(all(body == bodies[0] for body in bodies[1:]))
+
+
+class TestJobsTenantVisibility(unittest.TestCase):
+    @staticmethod
+    def _queue_record(owner=...):
+        extra_data: dict = {}
+        if owner is not ...:
+            extra_data["openclaw"] = {"tenant_id": owner}
+        return (1, "job-1", {"prompt": "secret"}, extra_data, ["1"])
+
+    @staticmethod
+    def _history_record(owner=...):
+        extra_data: dict = {}
+        if owner is not ...:
+            extra_data["openclaw"] = {"tenant_id": owner}
+        return {
+            "prompt": (1, "job-1", {"prompt": "secret"}, extra_data, ["1"]),
+            "outputs": {},
+        }
+
+    def test_single_tenant_keeps_native_unmarked_records(self):
+        security = _load_security_module(self)
+        record = self._queue_record()
+        result = security.filter_visible_job_records(
+            [record], source="queue", tenant_id="default", multi_tenant=False
+        )
+        self.assertEqual(result.records, (record,))
+        self.assertEqual(result.excluded_count, 0)
+
+    def test_multi_tenant_queue_matrix_is_exact_match_only(self):
+        security = _load_security_module(self)
+        exact = self._queue_record("team-a")
+        records = [
+            exact,
+            self._queue_record("team-b"),
+            self._queue_record(),
+            self._queue_record(["team-a"]),
+            ("malformed",),
+        ]
+        result = security.filter_visible_job_records(
+            records, source="queue", tenant_id="team-a", multi_tenant=True
+        )
+        self.assertEqual(result.records, (exact,))
+        self.assertEqual(result.excluded_count, 4)
+        self.assertEqual(result.malformed_count, 2)
+
+    def test_multi_tenant_history_matrix_is_exact_match_only(self):
+        security = _load_security_module(self)
+        exact = self._history_record("team-a")
+        records = [
+            exact,
+            self._history_record("team-b"),
+            self._history_record(),
+            {"prompt": "malformed"},
+        ]
+        result = security.filter_visible_job_records(
+            records, source="history", tenant_id="team-a", multi_tenant=True
+        )
+        self.assertEqual(result.records, (exact,))
+        self.assertEqual(result.excluded_count, 3)
+        self.assertEqual(result.malformed_count, 1)
+
+    def test_unknown_source_fails_closed(self):
+        security = _load_security_module(self)
+        with self.assertRaises(security.JobsSecurityError) as ctx:
+            security.filter_visible_job_records(
+                [], source="unknown", tenant_id="team-a", multi_tenant=True
+            )
+        self.assertEqual(ctx.exception.code, "jobs_source_invalid")
+
+    def test_registry_token_tenant_header_mismatch_fails_closed(self):
+        security = _load_security_module(self)
+        request = _request(headers={"X-OpenClaw-Tenant-Id": "team-b"})
+        token = SimpleNamespace(token_id="kid-test", tenant_id="team-a")
+        with patch.dict(os.environ, {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}):
+            with self.assertRaises(security.TenantBoundaryError) as ctx:
+                with security.jobs_request_tenant_scope(request, token):
+                    pass
+        self.assertEqual(ctx.exception.code, "tenant_mismatch")
+
+
+class TestJobsQueryContract(unittest.TestCase):
+    def test_defaults_match_frozen_contract(self):
+        security = _load_security_module(self)
+        self.assertTrue(
+            hasattr(security, "normalize_jobs_query"),
+            "S100 must expose bounded jobs query normalization",
+        )
+        query = security.normalize_jobs_query({})
+        self.assertIsNone(query.status)
+        self.assertIsNone(query.workflow_id)
+        self.assertEqual(query.sort_by, "created_at")
+        self.assertEqual(query.sort_order, "desc")
+        self.assertEqual(query.limit, 50)
+        self.assertEqual(query.offset, 0)
+        self.assertEqual(query.warnings, ())
+
+    def test_limit_offset_clamp_and_warnings_are_bounded(self):
+        security = _load_security_module(self)
+        self.assertTrue(hasattr(security, "normalize_jobs_query"))
+        query = security.normalize_jobs_query(
+            {"limit": "999999", "offset": "999999999999999999999999999"}
+        )
+        self.assertEqual(query.limit, 200)
+        self.assertEqual(query.offset, 10000)
+        warnings = query.to_pagination()["warnings"]
+        self.assertEqual(
+            {warning["code"] for warning in warnings},
+            {"R95_LIMIT_CLAMPED", "R95_OFFSET_CLAMPED"},
+        )
+        encoded = json.dumps(warnings, sort_keys=True)
+        self.assertNotIn("999999999999999999999999999", encoded)
+        self.assertNotIn('"raw"', encoded)
+
+        invalid = security.normalize_jobs_query({"limit": "not-an-int"})
+        self.assertEqual(invalid.limit, 50)
+        self.assertEqual(
+            invalid.to_pagination()["warnings"],
+            [
+                {
+                    "code": "R95_INVALID_LIMIT",
+                    "field": "limit",
+                    "normalized": 50,
+                }
+            ],
+        )
+
+    def test_invalid_enums_filters_and_unknown_fields_fail_closed(self):
+        security = _load_security_module(self)
+        self.assertTrue(hasattr(security, "normalize_jobs_query"))
+        cases = (
+            {"status": "unknown"},
+            {"sort_by": "priority"},
+            {"sort_order": "sideways"},
+            {"workflow_id": "x" * 129},
+            {"unexpected": "field"},
+        )
+        for payload in cases:
+            with self.subTest(field=next(iter(payload))):
+                with self.assertRaises(security.JobsSecurityError) as ctx:
+                    security.normalize_jobs_query(payload)
+                self.assertEqual(ctx.exception.code, "jobs_query_invalid")
+
+    def test_audit_detail_builder_is_content_free_for_all_outcomes(self):
+        security = _load_security_module(self)
+        self.assertTrue(
+            hasattr(security, "build_jobs_audit_details"),
+            "S100 must expose a content-free jobs audit detail builder",
+        )
+        for reason in (
+            "jobs_listed",
+            "jobs_admin_required",
+            "jobs_rate_limited",
+            "jobs_host_contract_unsupported",
+            "jobs_backend_unavailable",
+        ):
+            with self.subTest(reason=reason):
+                details = security.build_jobs_audit_details(
+                    reason,
+                    returned_count=5,
+                    excluded_count=2,
+                    malformed_count=1,
+                    ignored_secret="job-secret-id",
+                )
+                self.assertEqual(details["reason"], reason)
+                self.assertEqual(
+                    set(details),
+                    {
+                        "reason",
+                        "returned_count",
+                        "excluded_count",
+                        "malformed_count",
+                    },
+                )
+                self.assertNotIn("job-secret-id", json.dumps(details))
+
+        fallback = security.build_jobs_audit_details(
+            "hostile-secret-reason",
+            returned_count=10001,
+            excluded_count=-2,
+            malformed_count=True,
+            another_count="4",
+        )
+        self.assertEqual(
+            fallback,
+            {
+                "reason": "jobs_error",
+                "returned_count": 10000,
+                "excluded_count": 0,
+            },
+        )
+
+    def test_audit_emitter_supports_degraded_outcomes_without_content(self):
+        from api import routes
+
+        self.assertTrue(hasattr(routes, "_emit_jobs_list_audit"))
+        request = _request()
+        with patch.object(routes, "emit_audit_event") as audit:
+            routes._emit_jobs_list_audit(
+                request=request,
+                token_info=None,
+                outcome="unsupported",
+                status_code=501,
+                reason="jobs_host_contract_unsupported",
+                returned_count=0,
+                job_id="job-secret-id",
+            )
+            routes._emit_jobs_list_audit(
+                request=request,
+                token_info=None,
+                outcome="unexpected",
+                status_code=503,
+                reason="hostile-secret-reason",
+                payload="secret-payload",
+            )
+
+        unsupported, fallback = [call.kwargs for call in audit.call_args_list]
+        self.assertEqual(unsupported["outcome"], "unsupported")
+        self.assertEqual(
+            unsupported["details"],
+            {
+                "reason": "jobs_host_contract_unsupported",
+                "returned_count": 0,
+            },
+        )
+        self.assertEqual(fallback["outcome"], "error")
+        self.assertEqual(fallback["details"], {"reason": "jobs_error"})
+        self.assertNotIn(
+            "secret", json.dumps([unsupported["details"], fallback["details"]])
+        )
+
+
+class TestJobsProjectionPrivacy(unittest.TestCase):
+    def test_projection_is_allowlist_only_under_hostile_nested_input(self):
+        security = _load_security_module(self)
+        hostile = {
+            "id": "job-safe",
+            "status": "completed",
+            "priority": 1,
+            "create_time": 100,
+            "execution_start_time": 110,
+            "execution_end_time": 120,
+            "outputs_count": 2,
+            "workflow_id": "workflow-safe",
+            "prompt": {"text": "secret-prompt"},
+            "workflow": {"nodes": ["secret-workflow"]},
+            "extra_data": {"openclaw": {"tenant_id": "secret-tenant"}},
+            "preview_output": {"filename": "secret.png"},
+            "execution_error": {"traceback": "secret-traceback"},
+            "trace_id": "secret-trace",
+            "client_id": "secret-client",
+            "reasoning": {"thinking": "secret-reasoning"},
+            "internal": {"maintenance": "secret-internal"},
+        }
+        projected = security.project_job_summary(hostile)
+        self.assertEqual(
+            set(projected),
+            {
+                "id",
+                "status",
+                "priority",
+                "create_time",
+                "execution_start_time",
+                "execution_end_time",
+                "outputs_count",
+                "workflow_id",
+            },
+        )
+        encoded = json.dumps(projected, sort_keys=True)
+        for secret in (
+            "secret-prompt",
+            "secret-workflow",
+            "secret-tenant",
+            "secret.png",
+            "secret-traceback",
+            "secret-trace",
+            "secret-client",
+            "secret-reasoning",
+            "secret-internal",
+        ):
+            self.assertNotIn(secret, encoded)
+
+    def test_projection_rejects_invalid_required_and_oversized_fields(self):
+        security = _load_security_module(self)
+        cases = (
+            ("empty-id", {"id": "", "status": "pending"}, "jobs_record_invalid"),
+            (
+                "unknown-status",
+                {"id": "job", "status": "unknown"},
+                "jobs_record_invalid",
+            ),
+            (
+                "unhashable-status",
+                {"id": "job", "status": ["pending"]},
+                "jobs_record_invalid",
+            ),
+            (
+                "oversized-id",
+                {"id": "x" * 129, "status": "pending"},
+                "jobs_record_invalid",
+            ),
+            (
+                "huge-priority",
+                {"id": "job", "status": "pending", "priority": 2**4096},
+                "jobs_record_invalid",
+            ),
+            (
+                "negative-output-count",
+                {"id": "job", "status": "pending", "outputs_count": -1},
+                "jobs_record_invalid",
+            ),
+        )
+        for label, payload, code in cases:
+            with self.subTest(label=label):
+                try:
+                    security.project_job_summary(payload)
+                except security.JobsSecurityError as exc:
+                    self.assertEqual(exc.code, code)
+                except Exception as exc:  # pragma: no cover - RED diagnostic path
+                    self.fail(
+                        f"projection leaked {type(exc).__name__} instead of JobsSecurityError"
+                    )
+                else:
+                    self.fail("invalid projection unexpectedly succeeded")
+
+
+class TestJobsContractGovernance(unittest.TestCase):
+    def test_endpoint_metadata_is_admin(self):
+        from api.routes import jobs_handler
+
+        metadata = get_metadata(jobs_handler)
+        self.assertIsNotNone(metadata)
+        self.assertEqual(metadata.auth_tier, AuthTier.ADMIN)
+
+    def test_public_contract_and_generated_openapi_are_admin(self):
+        api_contract = (REPO_ROOT / "docs/release/api_contract.md").read_text(
+            encoding="utf-8"
+        )
+        self.assertIn(
+            "| `GET` | `/jobs` | `/moltbot/jobs` | Admin |",
+            api_contract,
+        )
+
+        openapi = (REPO_ROOT / "docs/openapi.yaml").read_text(encoding="utf-8")
+        jobs_block = openapi.split("  /jobs:\n", 1)[1].split("\n  /", 1)[0]
+        self.assertIn('x-openclaw-auth: "Admin"', jobs_block)
+        self.assertIn('x-openclaw-auth-tier: "admin"', jobs_block)
+        self.assertIn("OpenClawAdminToken:", jobs_block)
+        self.assertNotIn("OpenClawObservabilityToken:", jobs_block)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

tests/test_s33r76r78_auth_hardening.py

@@ -31,7 +31,10 @@ AUTH_CLASS_BY_ROUTE = {
     # high-sensitivity prompt/runtime context (S34). Keep this auth class in
     # sync with api/logs_tail.py to avoid accidental privilege regression.
     ("GET", "/logs/tail"): "admin",
-    ("GET", "/jobs"): "public-safe",
+    # IMPORTANT: jobs exposes cross-job operational metadata once R213 replaces
+    # the stub. Keep explicit runtime Admin enforcement aligned with metadata,
+    # OpenAPI, and the public API contract (S100).
+    ("GET", "/jobs"): "admin",
     # IMPORTANT:
     # Trace endpoint now returns high-sensitivity execution context and is
     # intentionally admin-only (S34). Keep as admin to prevent data leakage.