refactor(monitoring): N2 - tach monitoring_tab.py, CanonicalAuditLogger, MonitoringQueryService, go circular import, sandbox matrix

- ui/monitoring_tab.py (1546 dong) tach thanh presentation/monitoring/**
  (container + 7 tab/card + shared helper), ui/monitoring_tab.py con lai
  re-export shim de app.py khong doi.
- infrastructure/telemetry/audit_logger.py: CanonicalAuditLogger, core/audit_log.py
  thanh wrapper mong, tuong thich nguoc 100% voi schema .jsonl cu.
- application/monitoring/monitoring_query_service.py: MonitoringQueryService
  read-only, filter/sort/pagination, khong import PySide6.
- Go circular import model_pricing<->usage_tracker va agent_security<->
  agent_security_alert (core/agent_security_types.py moi).
- infrastructure/sandbox/sandbox_capabilities.py: SandboxCapabilityMatrix
  theo OS (Windows/Linux/macOS), chua dau noi vao core/sandbox_manager.py.
- conftest.py: sua loi checkout khong ten cowork_local khien pytest import
  nham thu muc khac.
- 77 test moi, 167/167 pass. QA da xac nhan UI/business logic khong doi
  (xem evidence/report/unified_report.html).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Hiep Ha Van
2026-08-25 23:52:36 +09:00
co-authored by Claude Sonnet 5
parent 86c27e2e79
commit 40b12ecb15
54 changed files with 3506 additions and 1637 deletions
+1
View File
@@ -0,0 +1 @@
"""application/ — Điều phối use-case. KHÔNG import PySide6. Gọi domain + interface hạ tầng."""
+1
View File
@@ -0,0 +1 @@
"""Application monitoring package: Monitoring query service for audit and metrics."""
@@ -0,0 +1,47 @@
"""Application-layer view of an audit event — decoupled from the
infrastructure ``CanonicalAuditEvent`` so ``application/`` doesn't need to
share a concrete class with ``infrastructure/`` (only the shape). Field names
match the canonical audit schema (see
``infrastructure/telemetry/audit_logger.py``) 1:1.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict
@dataclass(frozen=True)
class AuditEventDTO:
ts: str
kind: str
name: str
ok: bool
detail: str
agent_role: str = ""
account: str = ""
role: str = ""
machine: str = ""
@classmethod
def from_raw(cls, raw: Dict[str, Any]) -> "AuditEventDTO":
"""Tolerant of missing keys — accepts both a
``CanonicalAuditEvent.to_dict()`` result and any historical raw
``.jsonl`` row."""
return cls(
ts=str(raw.get("ts", "")),
kind=str(raw.get("kind", "")),
name=str(raw.get("name", "")),
ok=bool(raw.get("ok", False)),
detail=str(raw.get("detail", "")),
agent_role=str(raw.get("agent_role", "")),
account=str(raw.get("account", "")),
role=str(raw.get("role", "")),
machine=str(raw.get("machine", "")),
)
def to_dict(self) -> Dict[str, Any]:
return {
"ts": self.ts, "kind": self.kind, "agent_role": self.agent_role,
"name": self.name, "ok": self.ok, "detail": self.detail,
"account": self.account, "role": self.role, "machine": self.machine,
}
@@ -0,0 +1,54 @@
"""Read-only query service over audit events — filter + sort + pagination.
Pure Python: no PySide6 import, no UI code. Depends only on an injected
``AuditEventRepository`` (see ``repository/audit_event_repository.py``), so it
is fully unit-testable with ``InMemoryAuditEventRepository`` and independent
of file I/O or Qt.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import List, Optional
from .dto.audit_event_dto import AuditEventDTO
from .repository.audit_event_repository import AuditEventRepository
@dataclass(frozen=True)
class Page:
items: List[AuditEventDTO]
total: int
page: int
page_size: int
@property
def has_more(self) -> bool:
return self.page * self.page_size < self.total
class MonitoringQueryService:
"""Read-only. Callers ask for a filtered/sorted/paginated slice of the
audit log; this service never writes anything."""
def __init__(self, repository: AuditEventRepository) -> None:
self._repository = repository
def query(self, kind: Optional[str] = None, ok: Optional[bool] = None,
text: Optional[str] = None, sort_by: str = "ts",
descending: bool = True, page: int = 1, page_size: int = 50) -> Page:
events = self._repository.load(kind=kind)
if ok is not None:
events = [e for e in events if e.ok == ok]
if text:
needle = text.lower()
events = [e for e in events
if needle in e.name.lower() or needle in e.detail.lower()]
events = sorted(events, key=lambda e: getattr(e, sort_by, ""), reverse=descending)
total = len(events)
page = max(1, page)
start = (page - 1) * page_size
items = events[start:start + page_size] if page_size > 0 else events
return Page(items=items, total=total, page=page, page_size=page_size)
@@ -0,0 +1,41 @@
"""Audit-event repository — the boundary between ``MonitoringQueryService``
and where events actually live. ``CanonicalAuditEventRepository`` is the real
adapter (wraps an injected ``CanonicalAuditLogger``); ``InMemoryAuditEventRepository``
is a constructor-injected test double, following this repo's existing
``Fake*``/``Recording*`` convention (see ``tests/routing/*``,
``tests/test_project_context_mcp_template.py``) rather than ``unittest.mock``.
"""
from __future__ import annotations
from typing import List, Optional, Protocol
from ..dto.audit_event_dto import AuditEventDTO
class AuditEventRepository(Protocol):
def load(self, kind: Optional[str] = None) -> List[AuditEventDTO]:
...
class CanonicalAuditEventRepository:
"""Adapter over ``infrastructure.telemetry.audit_logger.CanonicalAuditLogger``
— the only place this application service reaches into infrastructure."""
def __init__(self, audit_logger) -> None:
self._audit_logger = audit_logger
def load(self, kind: Optional[str] = None) -> List[AuditEventDTO]:
events = self._audit_logger.load_events(kind=kind)
return [AuditEventDTO.from_raw(e.to_dict()) for e in events]
class InMemoryAuditEventRepository:
"""Test double — holds a fixed list of events, no file I/O."""
def __init__(self, events: List[AuditEventDTO]) -> None:
self._events = list(events)
def load(self, kind: Optional[str] = None) -> List[AuditEventDTO]:
if kind is None:
return list(self._events)
return [e for e in self._events if e.kind == kind]
+32
View File
@@ -0,0 +1,32 @@
"""Root pytest conftest — loaded before ``tests/conftest.py``.
This checkout lives on disk as ``Refactor`` (not ``cowork_local``), while
``tests/`` imports everything as ``from cowork_local... import ...`` and
``tests/conftest.py`` makes that resolve by putting this repo's *parent*
directory on ``sys.path`` (expecting the repo root itself to be named
``cowork_local``). A sibling folder literally named ``cowork_local`` (an
unrelated, older checkout) already exists next to this one, so without this
file Python would silently import THAT folder instead of this repository
whenever a test does ``import cowork_local``.
Registering the alias here — before ``tests/conftest.py`` touches
``sys.path`` — caches this repository in ``sys.modules['cowork_local']``
first, so the later ``sys.path`` mutation has nothing left to do (imports
are cached by name; the first successful import of a given name wins for
the rest of the process).
"""
from __future__ import annotations
import importlib.util
import sys
from pathlib import Path
_ROOT = Path(__file__).resolve().parent
if "cowork_local" not in sys.modules:
spec = importlib.util.spec_from_file_location(
"cowork_local", _ROOT / "__init__.py", submodule_search_locations=[str(_ROOT)],
)
module = importlib.util.module_from_spec(spec)
sys.modules["cowork_local"] = module
spec.loader.exec_module(module)
+2 -19
View File
@@ -27,27 +27,12 @@ from __future__ import annotations
import json import json
import re import re
from dataclasses import dataclass
from typing import List, Optional from typing import List, Optional
from ..providers.base import Provider from ..providers.base import Provider
from . import security_rules from . import security_rules
from .agent_security_alert import notify_admin
from .agent_security_types import SecurityBlocked, SecurityVerdict
class SecurityBlocked(RuntimeError):
"""A guardrail refused an action. ``verdict`` carries the full detail for
the admin alert; ``str(exc)`` is the short, user-facing reason."""
def __init__(self, verdict: "SecurityVerdict"):
super().__init__(verdict.reason or f"Blocked by agent security ({verdict.layer}).")
self.verdict = verdict
@dataclass
class SecurityVerdict:
allowed: bool
reason: str = ""
layer: str = "" # "prompt" | "attachment" | "command"
def combined_rules_text(config, max_chars: int = 8000, agent_kind: str = "cowork") -> str: def combined_rules_text(config, max_chars: int = 8000, agent_kind: str = "cowork") -> str:
@@ -236,7 +221,6 @@ def enforce_prompt(provider: Provider, messages: List[dict], config, emit,
emit({"type": "notice", "level": "warning", emit({"type": "notice", "level": "warning",
"text": f"🛡 Yêu cầu bị chặn bởi Agent Security: {verdict.reason}"}) "text": f"🛡 Yêu cầu bị chặn bởi Agent Security: {verdict.reason}"})
from . import audit_log from . import audit_log
from .agent_security_alert import notify_admin
audit_log.record("security_block", "prompt", False, verdict.reason) audit_log.record("security_block", "prompt", False, verdict.reason)
notify_admin(config, verdict, detail=user_text[:1000]) notify_admin(config, verdict, detail=user_text[:1000])
@@ -266,7 +250,6 @@ def enforce_command(provider: Provider, name: str, args: dict, config, emit,
emit({"type": "notice", "level": "warning", emit({"type": "notice", "level": "warning",
"text": f"🛡 Lệnh bị chặn bởi Agent Security ({verdict.layer}): {verdict.reason}"}) "text": f"🛡 Lệnh bị chặn bởi Agent Security ({verdict.layer}): {verdict.reason}"})
from . import audit_log from . import audit_log
from .agent_security_alert import notify_admin
audit_log.record("security_block", name, False, f"{verdict.layer}: {verdict.reason}") audit_log.record("security_block", name, False, f"{verdict.layer}: {verdict.reason}")
notify_admin(config, verdict, detail=command) notify_admin(config, verdict, detail=command)
+1 -1
View File
@@ -12,7 +12,7 @@ from __future__ import annotations
from typing import Tuple from typing import Tuple
from . import ms365_graph from . import ms365_graph
from .agent_security import SecurityVerdict from .agent_security_types import SecurityVerdict
from .ms365_auth import Ms365AuthError, get_access_token from .ms365_auth import Ms365AuthError, get_access_token
+33
View File
@@ -0,0 +1,33 @@
"""Shared value types for the Agent Security guardrails.
``SecurityVerdict``/``SecurityBlocked`` used to be defined in
``agent_security.py``, which forced ``agent_security_alert.py`` (which only
needs the *type*, to annotate/read ``notify_admin``'s ``verdict`` argument) to
import from it — while ``agent_security.py`` itself needed to call
``agent_security_alert.notify_admin()``, an architectural cycle only avoided
at runtime by deferring that second import inside a function body.
Hoisting the shared type into this dependency-free leaf module lets both
sides import it directly, so ``agent_security.py`` can import
``agent_security_alert`` at module top level too — no cycle, no deferred
imports needed for this pair.
"""
from __future__ import annotations
from dataclasses import dataclass
@dataclass
class SecurityVerdict:
allowed: bool
reason: str = ""
layer: str = "" # "prompt" | "attachment" | "command"
class SecurityBlocked(RuntimeError):
"""A guardrail refused an action. ``verdict`` carries the full detail for
the admin alert; ``str(exc)`` is the short, user-facing reason."""
def __init__(self, verdict: SecurityVerdict):
super().__init__(verdict.reason or f"Blocked by agent security ({verdict.layer}).")
self.verdict = verdict
+15 -72
View File
@@ -6,15 +6,23 @@ storage systems).
One JSON line per event, one file per day under ``~/.cowork_local/audit/`` — One JSON line per event, one file per day under ``~/.cowork_local/audit/`` —
same on-disk shape as ``usage_tracker.py`` (day-sharded ``.jsonl``, append-only, same on-disk shape as ``usage_tracker.py`` (day-sharded ``.jsonl``, append-only,
``record()`` never raises so audit logging can never break a chat turn). ``record()`` never raises so audit logging can never break a chat turn).
This module is now a thin, backward-compatible wrapper around
:class:`infrastructure.telemetry.audit_logger.CanonicalAuditLogger` — every
existing call site (``agent_security.py``, ``chat_agent.py``, ``tools.py``,
``ext_connectors.py``, ``mcp_client.py``, ``ms365_local.py``,
``permissions.py``, ``ui/structure_graph_view.py``, ``app.py``) keeps calling
``audit_log.set_identity``/``record``/``load_events`` exactly as before; only
the implementation moved.
""" """
from __future__ import annotations from __future__ import annotations
import json from datetime import date
from datetime import date, datetime
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
from ..config import CONFIG_DIR from ..config import CONFIG_DIR
from ..infrastructure.telemetry.audit_logger import CanonicalAuditLogger
AUDIT_DIR = CONFIG_DIR / "audit" AUDIT_DIR = CONFIG_DIR / "audit"
@@ -23,66 +31,21 @@ AUDIT_DIR = CONFIG_DIR / "audit"
# action), "mcp_call" (a call to an external MCP server's tool). # action), "mcp_call" (a call to an external MCP server's tool).
Kind = str Kind = str
# Process-global identity — who's logged in, their role, and this machine's _logger = CanonicalAuditLogger(AUDIT_DIR)
# name — set once right after login (app.py::run()), mirroring
# usage_tracker.py's identical pattern. NOT thread-local: fixed per process.
_identity_account = ""
_identity_role = ""
_identity_machine = ""
_identity_shared_dir = ""
def set_identity(account: str, machine: str, role: str = "", shared_dir: str = "") -> None: def set_identity(account: str, machine: str, role: str = "", shared_dir: str = "") -> None:
"""Called once after login succeeds. ``shared_dir``, when reachable, """Called once after login succeeds. ``shared_dir``, when reachable,
makes every subsequent :func:`record` ALSO best-effort-append to the makes every subsequent :func:`record` ALSO best-effort-append to the
shared cross-machine telemetry store (see :mod:`telemetry_shared`).""" shared cross-machine telemetry store (see :mod:`telemetry_shared`)."""
global _identity_account, _identity_role, _identity_machine, _identity_shared_dir _logger.set_identity(account, machine, role=role, shared_dir=shared_dir)
_identity_account = account or ""
_identity_role = role or ""
_identity_machine = machine or ""
_identity_shared_dir = shared_dir or ""
def record(kind: Kind, name: str, ok: bool, detail: str = "", def record(kind: Kind, name: str, ok: bool, detail: str = "",
agent_role: str = "") -> None: agent_role: str = "") -> None:
"""Append one audit event. Never raises — audit logging must never break """Append one audit event. Never raises — audit logging must never break
a chat turn, a permission decision, or a tool call.""" a chat turn, a permission decision, or a tool call."""
try: _logger.record(kind, name, ok, detail=detail, agent_role=agent_role)
now = datetime.now()
event = {
"ts": now.isoformat(timespec="seconds"),
"kind": kind,
"agent_role": agent_role or "",
"name": name or "",
"ok": bool(ok),
"detail": (detail or "")[:2000], # bounded — never let a huge blob bloat the log
"account": _identity_account,
"role": _identity_role,
"machine": _identity_machine,
}
AUDIT_DIR.mkdir(parents=True, exist_ok=True)
path = AUDIT_DIR / f"{now.strftime('%Y-%m-%d')}.jsonl"
with path.open("a", encoding="utf-8") as f:
f.write(json.dumps(event, ensure_ascii=False) + "\n")
_write_shared(event, now)
except Exception: # noqa: BLE001
pass
def _write_shared(event: Dict[str, Any], now: datetime) -> None:
"""Best-effort mirror of ``event`` into the shared cross-machine store —
one file PER MACHINE per day, so no two machines ever write the same
file. Never raises."""
if not _identity_shared_dir or not _identity_machine:
return
try:
shared = Path(_identity_shared_dir).expanduser() / "telemetry" / "audit"
shared.mkdir(parents=True, exist_ok=True)
path = shared / f"{_identity_machine}-{now.strftime('%Y-%m-%d')}.jsonl"
with path.open("a", encoding="utf-8") as f:
f.write(json.dumps(event, ensure_ascii=False) + "\n")
except Exception: # noqa: BLE001
pass
def load_events(start: Optional[date] = None, end: Optional[date] = None, def load_events(start: Optional[date] = None, end: Optional[date] = None,
@@ -91,25 +54,5 @@ def load_events(start: Optional[date] = None, end: Optional[date] = None,
"""Events between ``start``/``end`` (inclusive; None = unbounded), """Events between ``start``/``end`` (inclusive; None = unbounded),
optionally filtered to one ``kind`` — this IS how each Monitoring optionally filtered to one ``kind`` — this IS how each Monitoring
Dashboard panel gets its own slice of the same underlying log.""" Dashboard panel gets its own slice of the same underlying log."""
directory = directory or AUDIT_DIR events = _logger.load_events(start=start, end=end, kind=kind, directory=directory)
if not directory.exists(): return [e.to_dict() for e in events]
return []
events: List[Dict[str, Any]] = []
for path in sorted(directory.glob("*.jsonl")):
try:
day = datetime.strptime(path.stem, "%Y-%m-%d").date()
except ValueError:
continue
if (start and day < start) or (end and day > end):
continue
try:
for line in path.read_text(encoding="utf-8").splitlines():
if not line.strip():
continue
event = json.loads(line)
if kind is not None and event.get("kind") != kind:
continue
events.append(event)
except (OSError, json.JSONDecodeError):
continue
return events
+13 -3
View File
@@ -32,6 +32,14 @@ _SYMBOL_CCY = {"₫": "VND", "vnd": "VND", "đ": "VND",
_DEFAULT_UNIT = "Million tokens" _DEFAULT_UNIT = "Million tokens"
# Flat USD/1M-token fallback rates used by turn_cost_usd() when a model isn't
# in the price table. Owned here (not usage_tracker.DEFAULT_PRICING) so this
# module never needs to import usage_tracker — usage_tracker imports this
# module instead, keeping the dependency one-directional. Values match
# usage_tracker.DEFAULT_PRICING's price_per_mtok_in_usd/out_usd exactly.
_FALLBACK_RATE_IN_USD = 0.5
_FALLBACK_RATE_OUT_USD = 1.5
# ---- currency ------------------------------------------------------------ # ---- currency ------------------------------------------------------------
def _rates(config) -> Dict[str, float]: def _rates(config) -> Dict[str, float]:
@@ -138,9 +146,11 @@ def turn_cost_usd(model: str, in_tok: int, out_tok: int, config) -> float:
switches models (a different model → its own row / rates).""" switches models (a different model → its own row / rates)."""
rates = usd_rates_for(model, config) rates = usd_rates_for(model, config)
if rates is None: if rates is None:
from . import usage_tracker as ut usage = (getattr(config, "data", {}) or {}).get("usage") or {}
p = {**ut.DEFAULT_PRICING, **((getattr(config, "data", {}) or {}).get("usage") or {})} rates = {
rates = {"in": float(p["price_per_mtok_in_usd"]), "out": float(p["price_per_mtok_out_usd"])} "in": float(usage.get("price_per_mtok_in_usd", _FALLBACK_RATE_IN_USD)),
"out": float(usage.get("price_per_mtok_out_usd", _FALLBACK_RATE_OUT_USD)),
}
return (in_tok or 0) / 1e6 * rates["in"] + (out_tok or 0) / 1e6 * rates["out"] return (in_tok or 0) / 1e6 * rates["in"] + (out_tok or 0) / 1e6 * rates["out"]
+13 -1
View File
@@ -19,6 +19,7 @@ from pathlib import Path
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
from ..config import CONFIG_DIR from ..config import CONFIG_DIR
from . import model_pricing as mp
USAGE_DIR = CONFIG_DIR / "usage" USAGE_DIR = CONFIG_DIR / "usage"
@@ -49,6 +50,18 @@ def set_context(source: str, label: str = "") -> None:
_local.label = label _local.label = label
def current_context() -> tuple:
"""The ``(source, label)`` currently tagged on THIS thread.
Public counterpart to :func:`set_context`, added for
``infrastructure/telemetry/usage_sink.py``: a subscriber that needs to
attribute one event to a different surface must be able to save the
caller's context and put it back afterwards, instead of leaving the worker
thread permanently retagged.
"""
return getattr(_local, "source", "") or "", getattr(_local, "label", "") or ""
# ---- per-thread usage accumulator ----------------------------------------- # ---- per-thread usage accumulator -----------------------------------------
# A step/run that wants to know its OWN token/cost (not the all-time file total) # A step/run that wants to know its OWN token/cost (not the all-time file total)
# calls begin_accumulation(), reads accumulated() before/after a unit of work, # calls begin_accumulation(), reads accumulated() before/after a unit of work,
@@ -397,7 +410,6 @@ def set_budget(config, amount: float, currency: Optional[str] = None) -> None:
order and ``budget_set_at`` only has 1-second resolution, so a timestamp order and ``budget_set_at`` only has 1-second resolution, so a timestamp
cutoff could mis-include/exclude an event recorded in that same second — cutoff could mis-include/exclude an event recorded in that same second —
the count baseline is exact regardless of timing.""" the count baseline is exact regardless of timing."""
from . import model_pricing as mp
usage = config.data.setdefault("usage", {}) usage = config.data.setdefault("usage", {})
ccy = (currency or usage.get("currency") or "USD").upper() ccy = (currency or usage.get("currency") or "USD").upper()
usage["budget_amount_usd"] = mp.convert(float(amount or 0), ccy, "USD", config) usage["budget_amount_usd"] = mp.convert(float(amount or 0), ccy, "USD", config)
+1
View File
@@ -0,0 +1 @@
"""infrastructure/ — Chạm thế giới thật: file, keyring, HTTP, tiến trình. Cài đặt interface."""
+1
View File
@@ -0,0 +1 @@
"""Infrastructure sandbox package: OS-specific sandbox capability adapters."""
@@ -0,0 +1,179 @@
"""Sandbox capability matrix — which isolation backends exist on which OS,
and which one a given risk tier should prefer.
Pure policy/data: no subprocess execution, no PySide6, no dependency on
``core/sandbox_manager.py`` (that module owns the actual execution and isn't
in this task's editable scope — this matrix is a standalone, independently
testable module ready for that module's owner to wire in later).
The Windows entries mirror what ``core/sandbox_manager.py`` +
``core/appcontainer_sandbox.py``/``core/windows_sandbox_vm.py``/
``core/integrity_sandbox.py`` already implement today. Linux/macOS entries
are declared but marked ``implemented=False`` — today those platforms have no
real isolation backend (confirmed: ``core/appcontainer_sandbox.py`` and
``core/windows_sandbox_vm.py`` both hard-return ``False`` off Windows) — so
this matrix reports that honestly instead of pretending capabilities that
don't exist yet. Adding a real Linux/macOS backend later is a 1-line flip of
``implemented`` plus whatever backend module implements it; adding a whole
new OS is a call to :func:`register_profile`, no changes to
:class:`SandboxCapabilityMatrix` itself.
"""
from __future__ import annotations
import sys
from dataclasses import dataclass
from typing import Dict, Optional, Tuple
# Plain string constants (like core/audit_log.py's ``Kind``) rather than an
# Enum, so a brand-new OS can be registered without editing a closed type.
WINDOWS = "windows"
LINUX = "linux"
MACOS = "macos"
UNKNOWN = "unknown"
# Risk tiers — same vocabulary as security/command_risk_classifier.RiskLevel,
# kept as plain strings here so this module has zero dependency on the
# ``security/`` package (out of scope for this task).
SAFE = "SAFE"
MODERATE = "MODERATE"
HIGH = "HIGH"
CRITICAL = "CRITICAL"
BLOCKED = "blocked"
DIRECT = "direct"
def detect_os(platform_name: Optional[str] = None) -> str:
"""``platform_name`` defaults to ``sys.platform`` but can be injected for
testing (e.g. ``detect_os("linux")``, ``detect_os("darwin")``)."""
name = platform_name if platform_name is not None else sys.platform
if name.startswith("win"):
return WINDOWS
if name.startswith("linux"):
return LINUX
if name.startswith("darwin"):
return MACOS
return UNKNOWN
@dataclass(frozen=True)
class SandboxBackend:
name: str
isolation_level: str # "none" | "resource_limits" | "restricted_token" | "namespace" | "seatbelt" | "full_vm"
implemented: bool # whether a real backend exists today, vs. a declared placeholder
@dataclass(frozen=True)
class OsSandboxProfile:
operating_system: str
backends: Tuple[SandboxBackend, ...]
# risk tier -> ordered list of preferred backend names (first available wins)
routing: Dict[str, Tuple[str, ...]]
def _profile(operating_system: str, backends: Tuple[SandboxBackend, ...],
routing: Dict[str, Tuple[str, ...]]) -> OsSandboxProfile:
return OsSandboxProfile(operating_system=operating_system, backends=backends, routing=routing)
_WINDOWS_PROFILE = _profile(
WINDOWS,
backends=(
SandboxBackend(DIRECT, "none", True),
SandboxBackend("integrity_job_wfp", "resource_limits", True),
SandboxBackend("appcontainer", "restricted_token", True),
SandboxBackend("windows_sandbox", "full_vm", True),
),
routing={
SAFE: ("integrity_job_wfp", DIRECT),
MODERATE: ("integrity_job_wfp", DIRECT),
HIGH: ("appcontainer", "integrity_job_wfp"),
CRITICAL: ("windows_sandbox", "appcontainer", BLOCKED),
},
)
_LINUX_PROFILE = _profile(
LINUX,
backends=(
SandboxBackend(DIRECT, "none", True),
SandboxBackend("namespaces_bubblewrap", "namespace", False), # not implemented yet
),
routing={
SAFE: (DIRECT,),
MODERATE: (DIRECT,),
HIGH: ("namespaces_bubblewrap", BLOCKED),
CRITICAL: (BLOCKED,),
},
)
_MACOS_PROFILE = _profile(
MACOS,
backends=(
SandboxBackend(DIRECT, "none", True),
SandboxBackend("sandbox_exec", "seatbelt", False), # not implemented yet
),
routing={
SAFE: (DIRECT,),
MODERATE: (DIRECT,),
HIGH: ("sandbox_exec", BLOCKED),
CRITICAL: (BLOCKED,),
},
)
_UNKNOWN_PROFILE = _profile(
UNKNOWN,
backends=(),
routing={SAFE: (BLOCKED,), MODERATE: (BLOCKED,), HIGH: (BLOCKED,), CRITICAL: (BLOCKED,)},
)
_PROFILES: Dict[str, OsSandboxProfile] = {
WINDOWS: _WINDOWS_PROFILE,
LINUX: _LINUX_PROFILE,
MACOS: _MACOS_PROFILE,
UNKNOWN: _UNKNOWN_PROFILE,
}
def register_profile(profile: OsSandboxProfile) -> None:
"""Extension point for a brand-new OS: build an :class:`OsSandboxProfile`
and register it once — no change to :class:`SandboxCapabilityMatrix`
needed. Overwrites any existing profile for the same
``operating_system`` name (lets a caller override the built-in Windows/
Linux/macOS profiles too, e.g. once a real Linux backend ships)."""
_PROFILES[profile.operating_system] = profile
class SandboxCapabilityMatrix:
"""Answers, for one OS: which backends are actually available today, and
which one a given risk tier should prefer. Read-only policy — does not
execute anything."""
def __init__(self, operating_system: Optional[str] = None,
allow_direct_fallback: bool = True) -> None:
self.operating_system = operating_system if operating_system is not None else detect_os()
self._profile = _PROFILES.get(self.operating_system, _UNKNOWN_PROFILE)
self.allow_direct_fallback = allow_direct_fallback
def all_backends(self) -> Tuple[SandboxBackend, ...]:
"""Every backend declared for this OS, implemented or not."""
return self._profile.backends
def available_backends(self) -> Tuple[SandboxBackend, ...]:
"""Only backends with a real implementation today."""
return tuple(b for b in self._profile.backends if b.implemented)
def select_backend(self, risk_level: str) -> str:
"""The backend name to use for ``risk_level`` on this OS — the first
available (implemented) backend in that tier's preference order, else
``"direct"`` when allowed for a non-CRITICAL tier, else ``"blocked"``."""
available_names = {b.name for b in self.available_backends()}
preferred = self._profile.routing.get(risk_level.upper(), ())
for name in preferred:
if name == BLOCKED:
return BLOCKED
if name in available_names:
return name
if (self.allow_direct_fallback and DIRECT in available_names
and risk_level.upper() != CRITICAL):
return DIRECT
return BLOCKED
+1
View File
@@ -0,0 +1 @@
"""Infrastructure telemetry package: CanonicalAuditLogger and token usage sinks."""
+167
View File
@@ -0,0 +1,167 @@
"""Canonical audit event logging — the infrastructure behind
``core/audit_log.py``'s ``set_identity``/``record``/``load_events`` free
functions (kept as thin wrappers over a module-level singleton for backward
compatibility with every existing call site).
Same on-disk shape as before: one JSON line per event, one file per day
under ``~/.cowork_local/audit/`` (plus a best-effort mirror into a shared
cross-machine folder when an identity's ``shared_dir`` is set). ``record()``
never raises — audit logging must never break a chat turn, a permission
decision, or a tool call.
The event schema is unchanged (same field names, same order) so every
``.jsonl`` file written before this refactor remains fully readable. New
event kinds can be added by defining another ``KIND_*`` constant — nothing
about the schema itself needs to change to support one.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from datetime import date, datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
# Known kinds today. ``kind`` stays a plain str (not an enum) so a caller can
# always pass a new value without editing this module — these constants are
# just the documented, current vocabulary.
KIND_TOOL_CALL = "tool_call"
KIND_PERMISSION = "permission"
KIND_SECURITY_BLOCK = "security_block"
KIND_MCP_CALL = "mcp_call"
@dataclass(frozen=True)
class CanonicalAuditEvent:
"""One audit log entry. Field order matches the pre-refactor
``core/audit_log.py`` schema exactly, for byte-compatible JSON output."""
ts: str
kind: str
agent_role: str
name: str
ok: bool
detail: str
account: str
role: str
machine: str
def to_dict(self) -> Dict[str, Any]:
return {
"ts": self.ts,
"kind": self.kind,
"agent_role": self.agent_role,
"name": self.name,
"ok": self.ok,
"detail": self.detail,
"account": self.account,
"role": self.role,
"machine": self.machine,
}
@classmethod
def from_dict(cls, raw: Dict[str, Any]) -> "CanonicalAuditEvent":
"""Tolerant of missing keys, so old/partial rows never fail to load."""
return cls(
ts=str(raw.get("ts", "")),
kind=str(raw.get("kind", "")),
agent_role=str(raw.get("agent_role", "")),
name=str(raw.get("name", "")),
ok=bool(raw.get("ok", False)),
detail=str(raw.get("detail", "")),
account=str(raw.get("account", "")),
role=str(raw.get("role", "")),
machine=str(raw.get("machine", "")),
)
@dataclass
class _Identity:
account: str = ""
role: str = ""
machine: str = ""
shared_dir: str = ""
class CanonicalAuditLogger:
"""Day-sharded JSONL audit writer/reader. Process identity (who's logged
in, this machine's name) is set once via :meth:`set_identity`, mirroring
the pre-refactor module-global pattern but held as instance state so this
class can be constructed/injected instead of relying on globals."""
def __init__(self, audit_dir: Path):
self.audit_dir = Path(audit_dir)
self._identity = _Identity()
def set_identity(self, account: str, machine: str, role: str = "",
shared_dir: str = "") -> None:
"""Called once after login succeeds. ``shared_dir``, when reachable,
makes every subsequent :meth:`record` ALSO best-effort-append to the
shared cross-machine telemetry store."""
self._identity = _Identity(account=account or "", role=role or "",
machine=machine or "", shared_dir=shared_dir or "")
def record(self, kind: str, name: str, ok: bool, detail: str = "",
agent_role: str = "") -> None:
"""Append one audit event. Never raises."""
try:
now = datetime.now()
event = CanonicalAuditEvent(
ts=now.isoformat(timespec="seconds"),
kind=kind,
agent_role=agent_role or "",
name=name or "",
ok=bool(ok),
detail=(detail or "")[:2000],
account=self._identity.account,
role=self._identity.role,
machine=self._identity.machine,
)
self.audit_dir.mkdir(parents=True, exist_ok=True)
path = self.audit_dir / f"{now.strftime('%Y-%m-%d')}.jsonl"
with path.open("a", encoding="utf-8") as f:
f.write(json.dumps(event.to_dict(), ensure_ascii=False) + "\n")
self._write_shared(event, now)
except Exception: # noqa: BLE001
pass
def _write_shared(self, event: CanonicalAuditEvent, now: datetime) -> None:
identity = self._identity
if not identity.shared_dir or not identity.machine:
return
try:
shared = Path(identity.shared_dir).expanduser() / "telemetry" / "audit"
shared.mkdir(parents=True, exist_ok=True)
path = shared / f"{identity.machine}-{now.strftime('%Y-%m-%d')}.jsonl"
with path.open("a", encoding="utf-8") as f:
f.write(json.dumps(event.to_dict(), ensure_ascii=False) + "\n")
except Exception: # noqa: BLE001
pass
def load_events(self, start: Optional[date] = None, end: Optional[date] = None,
kind: Optional[str] = None,
directory: Optional[Path] = None) -> List[CanonicalAuditEvent]:
"""Events between ``start``/``end`` (inclusive; None = unbounded),
optionally filtered to one ``kind``."""
directory = directory or self.audit_dir
if not directory.exists():
return []
events: List[CanonicalAuditEvent] = []
for path in sorted(directory.glob("*.jsonl")):
try:
day = datetime.strptime(path.stem, "%Y-%m-%d").date()
except ValueError:
continue
if (start and day < start) or (end and day > end):
continue
try:
for line in path.read_text(encoding="utf-8").splitlines():
if not line.strip():
continue
raw = json.loads(line)
if kind is not None and raw.get("kind") != kind:
continue
events.append(CanonicalAuditEvent.from_dict(raw))
except (OSError, json.JSONDecodeError):
continue
return events
+1
View File
@@ -0,0 +1 @@
"""presentation/ — Widget Qt. Chỉ gọi xuống application, không gọi thẳng infrastructure."""
View File
+212
View File
@@ -0,0 +1,212 @@
"""Monitoring Dashboard — container only. Builds the tab strip, wires the
auto-refresh timer and language-change retranslation, and forwards nav-rail
sub-tab selection. Each sub-tab is its own class under ``tabs/``; this class
owns no display logic of its own beyond assembling and refreshing them.
Public API preserved exactly for ``app.py`` (which cannot be modified):
``MonitoringTab(ctx, cowork=None, structure=None, task_scheduler=None)``,
the ``status_message`` signal, ``select_subtab(index)``, ``nav_subtabs()``,
``hide_tab_bar()``.
"""
from __future__ import annotations
from typing import List
from PySide6.QtCore import QTimer, Signal
from PySide6.QtWidgets import QHBoxLayout, QLabel, QTabWidget, QVBoxLayout, QWidget
from ...core import audit_log
from ...i18n import on_language_changed, tr
from ...state import AppContext
from .tabs.action_logs_tab import ActionLogsTab
from .tabs.agent_status_tab import AgentStatusTab
from .tabs.mcp_tab import McpTab
from .tabs.overview_tab import OverviewTab
from .tabs.security_events_tab import SecurityEventsTab
_REFRESH_MS = 3000
# Comfortably larger than any realistic audit-log size — the event tables
# have never had pagination controls, so every tab still shows "all matching
# events" exactly like before; MonitoringQueryService's pagination support
# is exercised for real here, just not surfaced as UI (yet).
_UNBOUNDED_PAGE_SIZE = 100_000
class MonitoringTab(QWidget):
status_message = Signal(str)
def __init__(self, ctx: AppContext, cowork=None, structure=None, task_scheduler=None):
super().__init__()
self.ctx = ctx
self._cowork = cowork
self._structure = structure
self._task_scheduler = task_scheduler
root = QVBoxLayout(self)
head = QHBoxLayout()
self._title = QLabel()
self._title.setStyleSheet("font-weight:700; font-size:15px;")
head.addWidget(self._title)
head.addStretch(1)
root.addLayout(head)
self.tabs = QTabWidget()
root.addWidget(self.tabs, 1)
visible = self._tab_visible
self.overview_tab = OverviewTab(
ctx, on_status_message=self.status_message.emit,
on_settings_changed=self.refresh,
on_view_all_action_logs=self._show_action_logs_tab,
action_logs_tab_visible=visible("action_logs"))
self.tabs.addTab(self.overview_tab, "")
self.security_tab = SecurityEventsTab(ctx, on_refresh_all=self.refresh)
if visible("security_events"):
self.tabs.addTab(self.security_tab, "")
self.mcp_tab = McpTab(ctx, on_refresh_all=self.refresh)
if visible("mcp_history"):
self.tabs.addTab(self.mcp_tab, "")
self.action_tab = ActionLogsTab(ctx, on_refresh_all=self.refresh)
if visible("action_logs"):
self.tabs.addTab(self.action_tab, "")
self.status_tab = AgentStatusTab(
ctx, on_refresh_all=self.refresh,
cowork=cowork, structure=structure, task_scheduler=task_scheduler)
if visible("agent_status"):
self.tabs.addTab(self.status_tab, "")
# ---- Agents Admin (catalog: assign a role + pinned model per agent) --
from ...ui.agents_admin_tab import AgentsAdminTab
self.agents_admin_tab = AgentsAdminTab(ctx)
if visible("agents_admin"):
self.tabs.addTab(self.agents_admin_tab, "")
# ---- Tools (govern built-in tools + Connectors/MCP in one place) -----
from ...ui.tools_admin_tab import ToolsAdminTab
self.tools_admin_tab = ToolsAdminTab(ctx)
if visible("tools_admin"):
self.tabs.addTab(self.tools_admin_tab, "")
# ---- Icons (browse built-in icons + add custom icons for agents/flows) --
from ...ui.icons_admin_tab import IconsAdminTab
self.icons_admin_tab = IconsAdminTab(ctx)
self.tabs.addTab(self.icons_admin_tab, "")
self.tabs.setCurrentIndex(0)
self._timer = QTimer(self)
# Only the Overview cards auto-refresh on this tick — Security, MCP,
# Action Logs, Agent Status (and the admin tabs) are read-only tables
# a background re-sort would otherwise disturb mid-interaction; the
# user refreshes them explicitly via a "Refresh" button.
self._timer.setInterval(_REFRESH_MS)
self._timer.timeout.connect(self._auto_refresh)
self._timer.start()
on_language_changed(self._retranslate)
self.refresh()
# ---- nav integration: sub-tabs driven from the left nav rail ------------
def nav_subtabs(self):
"""(label, index, icon_name) for each sub-tab — the left nav lists
these as children under 'Monitoring'."""
by_widget = {self.agents_admin_tab: "robot", self.tools_admin_tab: "wrench",
self.icons_admin_tab: "star"}
for attr, name in (("security_tab", "shield"), ("mcp_tab", "plug"),
("action_tab", "bolt"), ("status_tab", "monitor")):
w = getattr(self, attr, None)
if w is not None:
by_widget[w] = name
out = []
for i in range(self.tabs.count()):
out.append((self.tabs.tabText(i), i, by_widget.get(self.tabs.widget(i), "dashboard")))
return out
def select_subtab(self, index: int) -> None:
if 0 <= index < self.tabs.count():
self.tabs.setCurrentIndex(index)
def hide_tab_bar(self) -> None:
"""Hide the in-content tab strip; the nav rail drives the sub-tabs."""
self.tabs.tabBar().hide()
def _show_action_logs_tab(self) -> None:
self.tabs.setCurrentIndex(self.tabs.indexOf(self.action_tab))
def _tab_visible(self, key: str) -> bool:
return self.ctx.role == "admin" or bool(self.ctx.config.monitoring_visibility.get(key, True))
def _set_tab_text_if_present(self, widget, text: str) -> None:
idx = self.tabs.indexOf(widget)
if idx >= 0:
self.tabs.setTabText(idx, text)
# ---- i18n ---------------------------------------------------------------
def _retranslate(self) -> None:
self._title.setText(tr("monitoring.title"))
if self.tabs.count():
self.tabs.setTabText(0, tr("monitoring.tab_overview"))
self._set_tab_text_if_present(self.security_tab, tr("monitoring.tab_security"))
self._set_tab_text_if_present(self.mcp_tab, tr("monitoring.tab_mcp"))
self._set_tab_text_if_present(self.action_tab, tr("monitoring.tab_actions"))
self._set_tab_text_if_present(self.status_tab, tr("monitoring.tab_agents"))
self._set_tab_text_if_present(self.agents_admin_tab, tr("monitoring.tab_agents_admin"))
self._set_tab_text_if_present(self.tools_admin_tab, tr("monitoring.tab_tools"))
self._set_tab_text_if_present(self.icons_admin_tab, tr("monitoring.tab_icons"))
self.overview_tab.retranslate()
self.security_tab.retranslate()
self.mcp_tab.retranslate()
self.action_tab.retranslate()
self.status_tab.retranslate()
self.refresh()
# ---- refresh -------------------------------------------------------------
def refresh(self) -> None:
"""Full refresh — Overview cards plus every table. Wired to the
top-of-page and per-section "Refresh" buttons, called once at
startup/language-change, but NOT to the auto-refresh timer (see
``_auto_refresh``)."""
events = self._load_events()
self._apply_events_to_event_tabs(events)
self.status_tab.refresh()
self.overview_tab.refresh(events)
def _auto_refresh(self) -> None:
"""3-second timer tick — Overview cards only (see ``refresh``)."""
self.overview_tab.refresh(self._load_events())
def _load_events(self) -> List[dict]:
shared_dir = self.ctx.config.shared_dir
if shared_dir:
from ...core import telemetry_shared
shared_events = telemetry_shared.load_shared_audit_events(shared_dir)
if shared_events:
return shared_events
return audit_log.load_events()
def _apply_events_to_event_tabs(self, events: List[dict]) -> None:
"""Filters the ALREADY-LOADED event list (see ``_load_events`` — one
shared local-or-shared decision per refresh, not one per tab) three
ways via :class:`MonitoringQueryService`, mirroring what each
``EventTable.set_events`` used to receive directly."""
from ...application.monitoring.dto.audit_event_dto import AuditEventDTO
from ...application.monitoring.monitoring_query_service import MonitoringQueryService
from ...application.monitoring.repository.audit_event_repository import (
InMemoryAuditEventRepository,
)
repository = InMemoryAuditEventRepository([AuditEventDTO.from_raw(e) for e in events])
service = MonitoringQueryService(repository)
def _events_for(kind) -> List[dict]:
page = service.query(kind=kind, page_size=_UNBOUNDED_PAGE_SIZE)
return [e.to_dict() for e in page.items]
self.security_tab.set_events(_events_for("security_block"))
self.mcp_tab.set_events(_events_for("mcp_call"))
self.action_tab.set_events(_events_for(None))
@@ -0,0 +1,45 @@
"""AI-assisted search-keyword filter — shared by the Security Events / MCP
Call History / Action Logs tabs' search box. Extracted verbatim from
``ui/monitoring_tab.py``'s ``MonitoringTab._ai_filter``.
Each caller owns its own ``state`` dict (just ``{}`` at construction) so a
repeat click is a no-op while a request is already in flight — mirrors the
original single ``self._ai_filter_worker`` attribute, without needing a
shared base class.
"""
from __future__ import annotations
from PySide6.QtWidgets import QLineEdit, QPushButton
def start_ai_filter(ctx, search: QLineEdit, ai_btn: QPushButton, state: dict) -> None:
query = search.text().strip()
if not query or state.get("worker") is not None:
return
ai_btn.setEnabled(False)
def job(worker):
provider = ctx.build_active_provider()
reply = provider.chat([
{"role": "system", "content":
"Turn the user's natural-language question about an audit/security event "
"log into ONE short search keyword. Reply with ONLY the keyword."},
{"role": "user", "content": query},
], cancel=worker.is_cancelled)
return {"keyword": (reply.get("content") or "").strip().splitlines()[0][:60]}
def done(result: dict) -> None:
state["worker"] = None
ai_btn.setEnabled(True)
search.setText(result.get("keyword") or query)
def failed(_err: str) -> None:
state["worker"] = None
ai_btn.setEnabled(True)
from ....core.worker import AgentWorker
w = AgentWorker(job)
w.finished_ok.connect(done)
w.failed.connect(failed)
state["worker"] = w
w.start()
+90
View File
@@ -0,0 +1,90 @@
"""Badge/label vocabulary shared by the Monitoring event tables and detail
panel — human labels for a raw audit ``name``, and the (QSS object name,
i18n key) pair for the "Trạng thái"/"Mức độ" pills.
``apply_badge`` replaces the two byte-identical
``_EventDetailPanel._apply_badge`` / ``MonitoringTab._set_badge`` static
methods the original file duplicated.
"""
from __future__ import annotations
from typing import Tuple
from PySide6.QtWidgets import QLabel
from ....i18n import tr
# Human label for the raw ``name`` an audit event is recorded under — the
# "Loại" field in the detail panel. Anything not in this map (custom tool
# names, etc.) just shows its raw name, same as the table's Hành động column.
_ACTION_LABEL_KEYS = {
"prompt": "monitoring.action_prompt",
"dangerous_command": "monitoring.action_dangerous_command",
"run_command": "monitoring.action_dangerous_command",
"install_package": "monitoring.action_install_package",
"path_outside_sandbox": "monitoring.action_path_outside_sandbox",
"network_blocked": "monitoring.action_network_blocked",
"secret_in_output": "monitoring.action_secret_in_output",
}
def action_label(name: str) -> str:
key = _ACTION_LABEL_KEYS.get(name)
return tr(key) if key else name
# (badge QSS object name, i18n key) for the "Trạng thái" pill, mapped onto
# the app's existing badge* tones (theme.py).
_STATUS_INFO = {
"path_outside_sandbox": ("badgeSuccess", "monitoring.status_path"),
"network_blocked": ("badge", "monitoring.status_network"),
"secret_in_output": ("badgeWarn", "monitoring.status_secret"),
}
_STATUS_DEFAULT = ("badgePurple", "monitoring.status_blocked")
def status_info(name: str, kind: str = "security_block", ok: bool = False) -> Tuple[str, str]:
if name in _STATUS_INFO:
return _STATUS_INFO[name]
if kind == "security_block":
# Security Events rows are always ok=False — an unmapped name here
# still means "blocked by some rule", never a plain failure.
return _STATUS_DEFAULT
# MCP calls / generic Action Logs rows: no fixed enforcement-rule
# vocabulary applies, so fall back to the event's own ok/fail outcome.
return ("badgeSuccess", "monitoring.status_ok") if ok else ("badgeDanger", "monitoring.status_failed")
def severity_info(name: str, kind: str = "security_block", ok: bool = False) -> Tuple[str, str]:
# An unapproved shell command is the one CRITICAL case; everything else
# blocked is MEDIUM.
if name in ("dangerous_command", "run_command"):
return "badgeDanger", "monitoring.severity_critical"
if kind == "security_block":
return "badgeWarn", "monitoring.severity_medium"
# A successful MCP call / action is routine (INFO); a failed one still
# deserves the same MEDIUM tone Security Events uses for a blocked rule.
return ("badge", "monitoring.severity_info") if ok else ("badgeWarn", "monitoring.severity_medium")
def agent_badge_name(name: str) -> str:
"""Badge tone for the Agent field's pill — the same identity-colour
mapping ``formatters.agent_avatar_colour`` uses, expressed as one of the
shared badge* QSS classes (theme.py) instead of a literal hex."""
if "Security" in name:
return "badgeDanger"
if "Cowork" in name:
return "badge"
if name == "schedule":
return "badgeWarn"
if name == "graphrag":
return "badgePurple"
if "Code" in name:
return "badgeSuccess"
return "badge"
def apply_badge(label: QLabel, object_name: str) -> None:
label.setObjectName(object_name)
label.style().unpolish(label)
label.style().polish(label)
@@ -0,0 +1,189 @@
"""Right-hand "Chi tiet su kien" detail panel shared by the Security Events /
MCP Call History / Action Logs tabs — the full record behind whichever row is
selected in an :class:`~.event_table.EventTable`. Extracted verbatim from
``ui/monitoring_tab.py``.
"""
from __future__ import annotations
from typing import Dict, List, Tuple
from PySide6.QtCore import Qt, QTimer, Signal
from PySide6.QtGui import QGuiApplication
from PySide6.QtWidgets import (
QHBoxLayout, QLabel, QPushButton, QScrollArea, QVBoxLayout, QWidget,
)
from ....core import agent_roles
from ....i18n import tr
from ....ui.icons import icon
from .badges import agent_badge_name, action_label, apply_badge, severity_info, status_info
from .formatters import event_id, fmt_event_time_full
from .layout_helpers import kv_row
# Cosmetic label only — no real policy-versioning system exists yet.
_STATIC_POLICY_LABEL = "security_policy_v2"
class EventDetailPanel(QWidget):
"""Laid out as three labelled sections, a terminal-style block quote for
the detail text, and a metadata footer, closed by the header close
button, the footer button, Esc, or a click outside the table/panel (see
:class:`~.event_table.ClickOutsideCloser`)."""
closed = Signal()
def __init__(self):
super().__init__()
self.setObjectName("monSection")
self._detail_text = ""
outer = QVBoxLayout(self)
hdr = QHBoxLayout()
self._title_lbl = QLabel()
self._title_lbl.setStyleSheet("font-weight:700;")
hdr.addWidget(self._title_lbl, 1)
self._close_btn = QPushButton()
self._close_btn.setIcon(icon("close"))
self._close_btn.setFlat(True)
self._close_btn.setFixedWidth(28)
self._close_btn.setCursor(Qt.PointingHandCursor)
self._close_btn.clicked.connect(self.closed.emit)
hdr.addWidget(self._close_btn)
outer.addLayout(hdr)
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll.setFrameShape(QScrollArea.NoFrame)
body = QWidget()
self._body_lay = QVBoxLayout(body)
self._body_lay.setContentsMargins(0, 0, 4, 0)
scroll.setWidget(body)
outer.addWidget(scroll, 1)
self._section_hdrs: List[Tuple[str, QLabel]] = []
self._rows: Dict[str, Tuple[QLabel, QLabel]] = {}
def _section(key: str) -> None:
lbl = QLabel()
lbl.setObjectName("detailSectionHdr")
self._body_lay.addWidget(lbl)
self._section_hdrs.append((key, lbl))
def _field(key: str) -> QLabel:
lbl, val = kv_row(self._body_lay)
self._rows[key] = (lbl, val)
return val
_section("general")
_field("time")
self._agent_val = _field("agent")
_field("account")
self._machine_val = _field("machine")
self._machine_val.setObjectName("monoChip")
_section("action")
self._type_val = _field("type")
self._type_val.setObjectName("neutralTag")
self._status_val = _field("status")
_section("block")
code_box = QWidget()
code_box.setObjectName("detailCodeBlock")
code_lay = QHBoxLayout(code_box)
code_lay.setContentsMargins(8, 6, 8, 6)
self._code_text = QLabel()
self._code_text.setObjectName("detailCodeText")
self._code_text.setWordWrap(True)
self._code_text.setTextInteractionFlags(Qt.TextSelectableByMouse)
code_lay.addWidget(self._code_text, 1)
self._copy_btn = QPushButton()
self._copy_btn.setObjectName("detailCopyBtn")
self._copy_btn.setCursor(Qt.PointingHandCursor)
self._copy_btn.clicked.connect(self._copy_detail)
code_lay.addWidget(self._copy_btn, 0, Qt.AlignVCenter)
self._body_lay.addWidget(code_box)
_section("metadata")
self._event_id_val = _field("event_id")
self._event_id_val.setObjectName("monoChip")
self._policy_val = _field("policy")
self._severity_val = _field("severity")
self._body_lay.addStretch(1)
footer = QHBoxLayout()
footer.setContentsMargins(0, 6, 0, 0)
self._footer_close_btn = QPushButton()
self._footer_close_btn.setObjectName("primary")
self._footer_close_btn.setCursor(Qt.PointingHandCursor)
self._footer_close_btn.clicked.connect(self.closed.emit)
footer.addWidget(self._footer_close_btn, 1)
outer.addLayout(footer)
def retranslate(self) -> None:
self._title_lbl.setText(tr("monitoring.security_detail_title"))
self._close_btn.setToolTip(tr("monitoring.security_detail_close"))
self._footer_close_btn.setText(tr("monitoring.security_detail_close"))
self._footer_close_btn.setIcon(icon("close"))
section_keys = {
"general": "monitoring.detail_section_general",
"action": "monitoring.detail_section_action",
"block": "monitoring.col_detail_block",
"metadata": "monitoring.detail_section_metadata",
}
for key, lbl in self._section_hdrs:
lbl.setText(tr(section_keys[key]).upper())
self._rows["time"][0].setText(tr("monitoring.col_time"))
self._rows["agent"][0].setText(tr("monitoring.col_agent"))
self._rows["account"][0].setText(tr("monitoring.col_account"))
self._rows["machine"][0].setText(tr("monitoring.col_machine"))
self._rows["type"][0].setText(tr("monitoring.detail_type"))
self._rows["status"][0].setText(tr("monitoring.detail_status"))
self._rows["event_id"][0].setText(tr("monitoring.detail_event_id"))
self._rows["policy"][0].setText(tr("monitoring.detail_policy"))
self._rows["severity"][0].setText(tr("monitoring.detail_severity"))
if not self._copy_btn.text() or self._copy_btn.text() != tr("monitoring.detail_copied"):
self._reset_copy_btn()
def show_event(self, ev: dict, row: int) -> None:
na = "—"
self._rows["time"][1].setText(fmt_event_time_full(ev.get("ts", "")) or na)
agent_label = agent_roles.label_for(ev.get("agent_role", "")) or na
self._agent_val.setText(agent_label)
apply_badge(self._agent_val,
agent_badge_name(agent_label) if agent_label != na else "badge")
self._rows["account"][1].setText(ev.get("account", "") or na)
self._machine_val.setText(ev.get("machine", "") or na)
name = ev.get("name", "")
kind = ev.get("kind", "security_block")
ok = ev.get("ok", False)
self._type_val.setText(action_label(name) or na)
status_badge, status_key = status_info(name, kind, ok)
self._status_val.setText(tr(status_key))
apply_badge(self._status_val, status_badge)
self._detail_text = ev.get("detail", "") or na
self._code_text.setText(self._detail_text)
self._reset_copy_btn()
self._event_id_val.setText(event_id(ev.get("ts", ""), row))
# The static policy label names a real enforcement ruleset — only
# meaningful for a Security Events row; MCP calls/generic actions
# were never evaluated against it.
self._policy_val.setText(_STATIC_POLICY_LABEL if kind == "security_block" else na)
severity_badge, severity_key = severity_info(name, kind, ok)
self._severity_val.setText(tr(severity_key))
apply_badge(self._severity_val, severity_badge)
def _copy_detail(self) -> None:
QGuiApplication.clipboard().setText(self._detail_text)
self._copy_btn.setText(tr("monitoring.detail_copied"))
self._copy_btn.setIcon(icon("check"))
QTimer.singleShot(1500, self._reset_copy_btn)
def _reset_copy_btn(self) -> None:
self._copy_btn.setText(tr("monitoring.detail_copy"))
self._copy_btn.setIcon(icon("document"))
@@ -0,0 +1,171 @@
"""Read-only audit-event table shared by the Security Events / MCP Call
History / Action Logs tabs, plus the click-outside-closes-detail-panel event
filter. Extracted verbatim from ``ui/monitoring_tab.py``.
"""
from __future__ import annotations
from typing import List, Optional
from PySide6.QtCore import QEvent, QObject, QRect, QSize, Qt
from PySide6.QtGui import QBrush, QColor
from PySide6.QtWidgets import QHeaderView, QTableWidget, QTableWidgetItem, QWidget
from ....core import agent_roles
from ....i18n import tr
from ....theme import current_palette
from ....ui.icons import DOT_GREEN, DOT_RED, icon
from .badges import action_label
from .formatters import agent_avatar_icon, fmt_event_time
_MAX_ROWS = 300
class _TimeItem(QTableWidgetItem):
"""The Time column shows "dd/MM hh:mm", which does not sort correctly as
text (day-of-month leads, not year/month) — so sorting compares the raw
ISO ``ts`` each item is built from instead of its displayed text."""
def __init__(self, raw_ts: str, display: str):
super().__init__(display)
self._raw_ts = raw_ts
def __lt__(self, other):
if isinstance(other, _TimeItem):
return self._raw_ts < other._raw_ts
return super().__lt__(other)
class EventTable(QTableWidget):
"""A read-only table of audit-log events — newest-first by default, and
every column header is click-to-sort (ascending/descending toggle; the
Time column sorts by the underlying ISO timestamp, not its "dd/MM hh:mm"
display text — see :class:`_TimeItem`)."""
# What each blocked action is, as a colour. Security events all record
# ok=False, so the tick/cross column said the same thing on every row; the
# useful distinction is WHICH rule fired.
_ACTION_TINTS = {
"prompt": "accent",
"dangerous_command": "danger",
"run_command": "danger",
"install_package": "warning",
"path_outside_sandbox": "success",
"network_blocked": "accent",
"secret_in_output": "warning",
}
def __init__(self, show_result: bool = True):
# Security Events drops the result column entirely (see _ACTION_TINTS).
self._show_result = show_result
super().__init__(0, 7 if show_result else 6)
self.setEditTriggers(QTableWidget.NoEditTriggers)
self.setSelectionBehavior(QTableWidget.SelectRows)
self.setIconSize(QSize(20, 20))
self.verticalHeader().setVisible(False)
# Fixed row height — letting Qt auto-size rows from content fought with
# the action column's cell widget geometry settling stale/oversized on
# an intermediate sizing pass, clipping the pill's text.
self.verticalHeader().setSectionResizeMode(QHeaderView.Fixed)
self.verticalHeader().setDefaultSectionSize(32)
self.setSortingEnabled(True)
header = self.horizontalHeader()
header.setStretchLastSection(True)
for col in range(self.columnCount() - 1):
header.setSectionResizeMode(col, QHeaderView.ResizeToContents)
def retranslate(self) -> None:
cols = [tr("monitoring.col_time"),
tr("monitoring.col_agent") if not self._show_result else tr("monitoring.col_role"),
tr("monitoring.col_account"), tr("monitoring.col_machine")]
if self._show_result:
cols += [tr("monitoring.col_name"), tr("monitoring.col_result")]
else:
cols += [tr("monitoring.col_action")]
cols += [tr("monitoring.col_detail_block") if not self._show_result else tr("monitoring.col_detail")]
self.setHorizontalHeaderLabels(cols)
def set_events(self, events: List[dict]) -> None:
events = sorted(events, key=lambda e: e.get("ts", ""), reverse=True)[:_MAX_ROWS]
self.setSortingEnabled(False)
self.setRowCount(len(events))
for row, ev in enumerate(events):
is_admin_violation = ev.get("role") == "admin" and not ev.get("ok", True)
cells = [
ev.get("ts", ""), agent_roles.label_for(ev.get("agent_role", "")),
ev.get("account", "") or "—", ev.get("machine", "") or "—",
ev.get("name", ""),
]
if self._show_result:
cells.append("")
cells.append((ev.get("detail") or "")[:300])
pal = current_palette()
for col, text in enumerate(cells):
item = (_TimeItem(str(text), fmt_event_time(str(text))) if col == 0
else QTableWidgetItem(str(text)))
if col == 0:
# Stash the full event (untruncated detail included) on the
# Time cell, so a click-to-open detail panel survives the
# user re-sorting the table by any column.
item.setData(Qt.UserRole, ev)
if col == 1:
item.setIcon(agent_avatar_icon(str(text)))
if self._show_result and col == 5:
item.setIcon(icon("check", color=DOT_GREEN) if ev.get("ok")
else icon("close", color=DOT_RED))
if not self._show_result and col == 4:
# Human-readable label, tinted by which rule fired, via the
# ITEM's own colours — NOT a setCellWidget() pill, which is
# pinned to a screen position rather than travelling with
# the item across a re-sort.
tint = getattr(pal, self._ACTION_TINTS.get(
ev.get("name", ""), "text_muted"), pal.text_muted)
item.setText(action_label(str(text)))
colour = QColor(tint)
item.setForeground(QBrush(colour))
soft = QColor(colour)
soft.setAlpha(38)
item.setBackground(QBrush(soft))
if is_admin_violation:
item.setBackground(QBrush(QColor(229, 72, 77, 60)))
self.setItem(row, col, item)
self.setSortingEnabled(True)
self.apply_filter(getattr(self, "_filter_needle", ""))
def apply_filter(self, needle: str) -> None:
self._filter_needle = (needle or "").strip().lower()
for row in range(self.rowCount()):
if not self._filter_needle:
self.setRowHidden(row, False)
continue
match = any(
self._filter_needle in (self.item(row, col).text().lower()
if self.item(row, col) else "")
for col in range(self.columnCount()))
self.setRowHidden(row, not match)
def event_at_row(self, row: int) -> Optional[dict]:
item = self.item(row, 0)
return item.data(Qt.UserRole) if item else None
class ClickOutsideCloser(QObject):
"""Closes the event-detail panel on a click anywhere outside the
table/panel splitter — judged by screen-space geometry (is the click's
global position inside the splitter's on-screen rectangle), not by which
exact widget object received the event (unreliable mid-drag on the
splitter's handle)."""
def __init__(self, table: "EventTable", panel: QWidget, container: QWidget):
super().__init__(container)
self._table = table
self._panel = panel
self._container = container
def eventFilter(self, obj, event) -> bool:
if event.type() == QEvent.MouseButtonPress and self._panel.isVisible():
global_pos = event.globalPosition().toPoint()
top_left = self._container.mapToGlobal(self._container.rect().topLeft())
rect = QRect(top_left, self._container.size())
if not rect.contains(global_pos):
self._table.clearSelection()
return False
@@ -0,0 +1,115 @@
"""Page scaffolding shared by the Security Events / MCP Call History / Action
Logs / Agent Status tabs: title + refresh button, optional search box +
AI-filter button, optional table+detail-panel split. Extracted from
``ui/monitoring_tab.py``'s ``MonitoringTab._wrap_with_filter``/
``_sync_event_detail`` — those used to be rebuilt 3 times almost identically
for the 3 event-table pages; this is the one shared implementation.
Builds directly into a caller-supplied ``page`` widget (which must not have a
layout yet) and returns a dict of the sub-widgets the caller needs to keep
(e.g. to implement its own ``retranslate()``).
"""
from __future__ import annotations
from typing import Callable, Dict, Optional
from PySide6.QtCore import Qt
from PySide6.QtGui import QKeySequence, QShortcut
from PySide6.QtWidgets import (
QApplication, QHBoxLayout, QLabel, QLineEdit, QPushButton, QSplitter,
QTableWidget, QVBoxLayout, QWidget,
)
from ....i18n import tr
from ....ui.icons import icon
from .event_table import ClickOutsideCloser, EventTable
from .event_detail_panel import EventDetailPanel
def _sync_event_detail(table: EventTable, panel: EventDetailPanel) -> None:
# currentRow() alone is not enough: clearSelection() (used by the panel's
# close button) drops the selection but leaves the current cell in
# place, so a stale currentRow() would keep the panel open.
row = table.currentRow()
has_selection = bool(table.selectedItems())
ev = table.event_at_row(row) if (has_selection and row >= 0) else None
if ev:
panel.show_event(ev, row)
panel.setVisible(bool(ev))
def build_filter_scaffold(
page: QWidget, table: QTableWidget, *, on_refresh: Callable[[], None],
title_key: Optional[str] = None, with_search: bool = True,
with_detail: bool = False,
on_ai_filter: Optional[Callable[[QLineEdit, QPushButton], None]] = None,
) -> Dict[str, object]:
lay = QVBoxLayout(page)
lay.setContentsMargins(0, 0, 0, 0)
parts: Dict[str, object] = {}
if title_key:
hdr = QHBoxLayout()
title_lbl = QLabel(tr(title_key))
title_lbl.setStyleSheet("font-weight:700; font-size:14px;")
hdr.addWidget(title_lbl)
hdr.addStretch(1)
refresh_btn = QPushButton(tr("monitoring.refresh"))
refresh_btn.setIcon(icon("refresh"))
refresh_btn.setObjectName("primary")
refresh_btn.setCursor(Qt.PointingHandCursor)
refresh_btn.clicked.connect(on_refresh)
hdr.addWidget(refresh_btn)
lay.addLayout(hdr)
parts.update(title_lbl=title_lbl, title_key=title_key, title_refresh_btn=refresh_btn)
if with_search:
row = QHBoxLayout()
search = QLineEdit()
search.setPlaceholderText(tr("monitoring.filter_placeholder"))
search.textChanged.connect(table.apply_filter)
ai_btn = QPushButton(tr("monitoring.ai_filter_btn"))
ai_btn.setIcon(icon("sparkle"))
ai_btn.setToolTip(tr("monitoring.ai_filter_tooltip"))
ai_btn.setCursor(Qt.PointingHandCursor)
if on_ai_filter is not None:
ai_btn.clicked.connect(lambda: on_ai_filter(search, ai_btn))
row.addWidget(search, 1)
row.addWidget(ai_btn)
lay.addLayout(row)
parts.update(filter_edit=search, ai_filter_btn=ai_btn)
if with_detail:
# Click a row -> its full record opens in a detail panel on the
# right (the table itself clips the detail text to 300 chars). A
# pointing-hand cursor over the rows signals that they're clickable.
table.setCursor(Qt.PointingHandCursor)
detail = EventDetailPanel()
detail.setVisible(False)
detail.closed.connect(table.clearSelection)
table.itemSelectionChanged.connect(lambda: _sync_event_detail(table, detail))
split = QSplitter(Qt.Horizontal)
split.addWidget(table)
split.addWidget(detail)
split.setStretchFactor(0, 1)
split.setStretchFactor(1, 0)
split.setChildrenCollapsible(False)
split.setSizes([700, 320])
lay.addWidget(split, 1)
parts["detail_panel"] = detail
# Esc, anywhere focus is inside this page, closes the panel the same
# way the close button does.
esc = QShortcut(QKeySequence(Qt.Key_Escape), page)
esc.setContext(Qt.WidgetWithChildrenShortcut)
esc.activated.connect(table.clearSelection)
parts["detail_esc_shortcut"] = esc
# A click outside both the table and the panel also closes it.
click_filter = ClickOutsideCloser(table, detail, split)
QApplication.instance().installEventFilter(click_filter)
parts["detail_click_filter"] = click_filter
else:
lay.addWidget(table, 1)
return parts
@@ -0,0 +1,113 @@
"""Formatting helpers shared by the Monitoring tabs — timestamps, byte
counts, and the per-agent colour-coded initials avatar.
Extracted from ``ui/monitoring_tab.py`` verbatim (same output for the same
input); the three timestamp formatters used to each repeat their own
``datetime.fromisoformat`` + ``try/except (TypeError, ValueError)`` guard —
that parse step is now a single shared ``_parse_iso`` helper.
"""
from __future__ import annotations
from datetime import datetime
from typing import Optional
from PySide6.QtCore import Qt
from PySide6.QtGui import QColor, QFont, QIcon, QPainter, QPixmap
from ....i18n import tr
def _parse_iso(ts: str) -> Optional[datetime]:
try:
return datetime.fromisoformat(ts)
except (TypeError, ValueError):
return None
def fmt_bytes(n: float) -> str:
for unit in ("B", "KB", "MB", "GB"):
if n < 1024:
return f"{n:.0f} {unit}"
n /= 1024
return f"{n:.1f} TB"
def fmt_event_time(ts: str) -> str:
""""dd/MM hh:mm" for the Time column — e.g. 25/05 15:03."""
dt = _parse_iso(ts)
return dt.strftime("%d/%m %H:%M") if dt else ts
_MIDDLE_DOT = chr(0xB7)
def fmt_event_time_full(ts: str) -> str:
""""dd/MM/yyyy [middle dot] HH:mm:ss" — the detail panel's Thoi gian field."""
dt = _parse_iso(ts)
return dt.strftime(f"%d/%m/%Y {_MIDDLE_DOT} %H:%M:%S") if dt else ts
def relative_time(ts: str) -> str:
"""A short "Xm ago"-style string for an audit-log ``ts``; "" if
unparsable."""
dt = _parse_iso(ts)
if dt is None:
return ""
delta = (datetime.now() - dt).total_seconds()
if delta < 60:
return tr("monitoring.time_just_now")
if delta < 3600:
return tr("monitoring.time_minutes_ago", n=int(delta // 60))
if delta < 86400:
return tr("monitoring.time_hours_ago", n=int(delta // 3600))
return tr("monitoring.time_days_ago", n=int(delta // 86400))
def event_id(ts: str, row: int) -> str:
"""A display-only id in the ``evt_<timestamp digits>_<row>`` shape the
mockup uses — the real audit log has no native event id, so this is
derived from the timestamp and the row's position in the currently
displayed (sorted) table, not persisted anywhere."""
digits = "".join(ch for ch in ts if ch.isdigit())[:12]
return f"evt_{digits}_{row:03d}"
def agent_initials(name: str) -> str:
"""First letter of each word, max 2."""
return "".join(w[0] for w in name.split() if w)[:2].upper()
def agent_avatar_colour(name: str) -> str:
"""A fixed identity colour per agent kind, unchanged by theme."""
if "Security" in name:
return "#D13438"
if "Cowork" in name:
return "#0078D4"
if name == "schedule" or "Task" in name:
return "#FFB900"
if name == "graphrag" or "Knowledge" in name:
return "#8764B8"
if "Code" in name:
return "#107C10"
if "Planner" in name or "Reasoning" in name:
return "#8A8886"
return "#0078D4"
def agent_avatar_icon(name: str, size: int = 20) -> QIcon:
"""A small round initials badge for the Agent column."""
pm = QPixmap(size, size)
pm.fill(Qt.transparent)
p = QPainter(pm)
p.setRenderHint(QPainter.Antialiasing)
p.setPen(Qt.NoPen)
p.setBrush(QColor(agent_avatar_colour(name)))
p.drawEllipse(0, 0, size, size)
font = QFont()
font.setPixelSize(max(7, size // 2))
font.setBold(True)
p.setFont(font)
p.setPen(QColor("#FFFFFF"))
p.drawText(pm.rect(), Qt.AlignCenter, agent_initials(name))
p.end()
return QIcon(pm)
@@ -0,0 +1,27 @@
"""``kv_row`` — the "label — stretch — value" row shape that
``ui/monitoring_tab.py`` used to redefine as 3 byte-identical local closures
(Sandbox Details' ``_kv``, Permissions' ``_pkv``, the detail panel's
``_field``). Overview's Resource Usage row (``_pair``) is a genuinely
different layout (inline on one horizontal line with "·" separators, no
stretch) and is intentionally NOT folded into this helper — unifying it would
risk a visible layout change.
"""
from __future__ import annotations
from typing import Tuple
from PySide6.QtWidgets import QHBoxLayout, QLabel, QVBoxLayout
def kv_row(outer_layout: QVBoxLayout, hint_object_name: str = "hint") -> Tuple[QLabel, QLabel]:
"""Appends a new ``label — stretch — value`` row to ``outer_layout``.
Returns ``(label, value)``."""
row = QHBoxLayout()
label = QLabel()
label.setObjectName(hint_object_name)
value = QLabel()
row.addWidget(label)
row.addStretch(1)
row.addWidget(value)
outer_layout.addLayout(row)
return label, value
@@ -0,0 +1,17 @@
"""Opens the Settings dialog and runs a callback afterward — shared by the
Sandbox Details and Permissions cards' "Edit" buttons, both of which used to
call the identical ``MonitoringTab._open_settings_and_refresh``.
"""
from __future__ import annotations
from typing import Callable
from PySide6.QtWidgets import QWidget
def open_settings_and_notify(ctx, parent: QWidget, on_changed: Callable[[], None]) -> None:
from ....ui.settings_dialog import SettingsDialog
dlg = SettingsDialog(ctx, parent)
dlg.exec()
on_changed()
@@ -0,0 +1,45 @@
"""Action Logs tab — the full audit log, newest first. Extracted from
``ui/monitoring_tab.py``'s action-table wiring inside
``MonitoringTab.__init__``.
"""
from __future__ import annotations
from typing import Callable, List
from PySide6.QtWidgets import QLineEdit, QPushButton, QWidget
from ....i18n import tr
from ..shared.ai_filter import start_ai_filter
from ..shared.event_table import EventTable
from ..shared.filter_scaffold import build_filter_scaffold
class ActionLogsTab(QWidget):
def __init__(self, ctx, on_refresh_all: Callable[[], None]):
super().__init__()
self._ctx = ctx
self._ai_state: dict = {}
self.table = EventTable()
parts = build_filter_scaffold(
self, self.table, on_refresh=on_refresh_all,
title_key="monitoring.action_logs_title",
with_search=True, with_detail=True, on_ai_filter=self._start_ai_filter)
self.title_lbl = parts["title_lbl"]
self.title_key = parts["title_key"]
self.title_refresh_btn = parts["title_refresh_btn"]
self.filter_edit = parts["filter_edit"]
self.ai_filter_btn = parts["ai_filter_btn"]
self.detail_panel = parts["detail_panel"]
def set_events(self, events: List[dict]) -> None:
self.table.set_events(events)
def retranslate(self) -> None:
self.table.retranslate()
self.filter_edit.setPlaceholderText(tr("monitoring.filter_placeholder"))
self.detail_panel.retranslate()
self.title_lbl.setText(tr(self.title_key))
self.title_refresh_btn.setText(tr("monitoring.refresh"))
def _start_ai_filter(self, search: QLineEdit, ai_btn: QPushButton) -> None:
start_ai_filter(self._ctx, search, ai_btn, self._ai_state)
@@ -0,0 +1,92 @@
"""Agent Status tab — which agent roles are currently running, read from the
existing ``ChatPanel``/``TaskScheduler``/GraphRAG-ask-worker state (no new
runtime tracking of its own). Extracted from ``ui/monitoring_tab.py``'s
status-table wiring + ``_refresh_agent_status``.
"""
from __future__ import annotations
from typing import Callable, Optional
from PySide6.QtCore import QSize
from PySide6.QtWidgets import QHeaderView, QTableWidget, QTableWidgetItem, QWidget
from ....core import agent_roles
from ....i18n import tr
from ....ui.widgets import badge_pill_widget
from ..shared.filter_scaffold import build_filter_scaffold
from ..shared.formatters import agent_avatar_icon
class AgentStatusTab(QWidget):
def __init__(self, ctx, on_refresh_all: Callable[[], None],
cowork=None, structure=None, task_scheduler=None):
super().__init__()
self._ctx = ctx
self._cowork = cowork
self._structure = structure
self._task_scheduler = task_scheduler
self.table = QTableWidget(0, 3)
self.table.setEditTriggers(QTableWidget.NoEditTriggers)
# No detail to open on click, no filtering — a plain read-only table,
# so selection is off rather than left dangling with no effect.
self.table.setSelectionMode(QTableWidget.NoSelection)
self.table.verticalHeader().setVisible(False)
self.table.horizontalHeader().setStretchLastSection(True)
self.table.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeToContents)
self.table.setColumnWidth(1, 130) # status is a cell widget — size it explicitly
self.table.setIconSize(QSize(20, 20))
parts = build_filter_scaffold(
self, self.table, on_refresh=on_refresh_all,
title_key="monitoring.agent_status_title", with_search=False, with_detail=False)
self.title_lbl = parts["title_lbl"]
self.title_key = parts["title_key"]
self.title_refresh_btn = parts["title_refresh_btn"]
def retranslate(self) -> None:
self.title_lbl.setText(tr(self.title_key))
self.title_refresh_btn.setText(tr("monitoring.refresh"))
self.table.setHorizontalHeaderLabels([
tr("monitoring.col_agent"), tr("monitoring.detail_status"), tr("monitoring.col_source"),
])
def refresh(self) -> None:
cowork_n = len(self._cowork.active_workers()) if self._cowork is not None else 0
task_n = self._task_scheduler.running_count() if self._task_scheduler is not None else 0
ask_worker = getattr(self._structure, "_ask_worker", None)
knowledge_n = 1 if (ask_worker is not None and ask_worker.isRunning()) else 0
# Security is a system-management agent that runs INLINE on the
# active turn (agent_security prompt/command validation) — there is
# no separate worker to count, so its "active" cell shows On/Off
# from Settings instead of a live count.
sec_on = bool(self._ctx.config.agent_security.get("enabled"))
rows = [
(agent_roles.COWORK, cowork_n, tr("monitoring.source_cowork")),
(agent_roles.TASK, task_n, tr("monitoring.source_task")),
(agent_roles.KNOWLEDGE, knowledge_n, tr("monitoring.source_knowledge")),
(agent_roles.PLANNER, None, tr("monitoring.source_planner")),
(agent_roles.REASONING, None, tr("monitoring.source_reasoning")),
(agent_roles.SECURITY, None, tr("monitoring.source_security")),
]
self.table.setRowCount(len(rows))
for row, (role_key, count, source) in enumerate(rows):
label = agent_roles.label_for(role_key)
name_item = QTableWidgetItem(label)
name_item.setIcon(agent_avatar_icon(label))
self.table.setItem(row, 0, name_item)
if role_key == agent_roles.SECURITY:
running = sec_on
status_text = tr("monitoring.on") if sec_on else tr("monitoring.off")
elif count is not None:
running = count > 0
status_text = tr("monitoring.active_n", n=count) if running else tr("monitoring.idle")
else:
running = False
status_text = "—"
badge_tone = "badgeSuccess" if running else "badgeNeutral"
self.table.setCellWidget(row, 1, badge_pill_widget(status_text, badge_tone))
self.table.setItem(row, 2, QTableWidgetItem(source))
+45
View File
@@ -0,0 +1,45 @@
"""MCP Call History tab — the audit log filtered to ``kind="mcp_call"``.
Extracted from ``ui/monitoring_tab.py``'s MCP-table wiring inside
``MonitoringTab.__init__``.
"""
from __future__ import annotations
from typing import Callable, List
from PySide6.QtWidgets import QLineEdit, QPushButton, QWidget
from ....i18n import tr
from ..shared.ai_filter import start_ai_filter
from ..shared.event_table import EventTable
from ..shared.filter_scaffold import build_filter_scaffold
class McpTab(QWidget):
def __init__(self, ctx, on_refresh_all: Callable[[], None]):
super().__init__()
self._ctx = ctx
self._ai_state: dict = {}
self.table = EventTable()
parts = build_filter_scaffold(
self, self.table, on_refresh=on_refresh_all,
title_key="monitoring.mcp_history_title",
with_search=True, with_detail=True, on_ai_filter=self._start_ai_filter)
self.title_lbl = parts["title_lbl"]
self.title_key = parts["title_key"]
self.title_refresh_btn = parts["title_refresh_btn"]
self.filter_edit = parts["filter_edit"]
self.ai_filter_btn = parts["ai_filter_btn"]
self.detail_panel = parts["detail_panel"]
def set_events(self, events: List[dict]) -> None:
self.table.set_events(events)
def retranslate(self) -> None:
self.table.retranslate()
self.filter_edit.setPlaceholderText(tr("monitoring.filter_placeholder"))
self.detail_panel.retranslate()
self.title_lbl.setText(tr(self.title_key))
self.title_refresh_btn.setText(tr("monitoring.refresh"))
def _start_ai_filter(self, search: QLineEdit, ai_btn: QPushButton) -> None:
start_ai_filter(self._ctx, search, ai_btn, self._ai_state)
@@ -0,0 +1,313 @@
"""Overview tab — the card-based dashboard (Token Usage & Cost, Resource
Usage, Recent Activity, Sandbox Details + nested Permissions, Model Pricing,
Audit Log preview). Extracted from ``ui/monitoring_tab.py``'s
``_build_overview_page`` and the refresh/pricing/budget methods it wires to.
"""
from __future__ import annotations
import time
from typing import Callable, List
from PySide6.QtCore import Qt
from PySide6.QtWidgets import (
QGridLayout, QGroupBox, QHBoxLayout, QLabel, QProgressBar, QPushButton,
QScrollArea, QVBoxLayout, QWidget,
)
from ....core import usage_tracker as ut
from ....i18n import tr
from ....theme import current_palette
from ....ui.icons import DOT_AMBER, DOT_GREEN, DOT_RED, icon
from ....ui.widgets import BudgetCard, StatCard, fmt_tokens
from ..shared.formatters import fmt_bytes, relative_time
from .pricing_panel import PricingPanel
from .sandbox_tab import SandboxDetailsCard
class OverviewTab(QWidget):
def __init__(self, ctx, *, on_status_message: Callable[[str], None],
on_settings_changed: Callable[[], None],
on_view_all_action_logs: Callable[[], None],
action_logs_tab_visible: bool):
super().__init__()
self.ctx = ctx
self._on_status_message = on_status_message
self._on_view_all_action_logs = on_view_all_action_logs
self._last_io_sample = None
self._res_first = True
outer = QVBoxLayout(self)
outer.setContentsMargins(0, 0, 0, 0)
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll.setFrameShape(QScrollArea.NoFrame)
content = QWidget()
scroll.setWidget(content)
outer.addWidget(scroll)
# ONE main column, scrolled vertically, sections in a fixed order.
root = QVBoxLayout(content)
root.setSpacing(12)
self._build_usage_section(root)
self._build_activity_section() # added to `root` further down, beside the audit log
self._build_resource_section(root)
self.sandbox_card = SandboxDetailsCard(ctx, on_settings_changed)
root.addWidget(self.sandbox_card)
self.pricing_panel = PricingPanel(ctx, on_status_message)
root.addWidget(self.pricing_panel)
root.addWidget(self.activity_group)
self._build_audit_section(root, action_logs_tab_visible)
root.addStretch(1)
# ---- Token Usage & Cost ------------------------------------------------
def _build_usage_section(self, root: QVBoxLayout) -> None:
self.usage_group = QGroupBox()
self.usage_group.setObjectName("monSection")
usage_lay = QGridLayout(self.usage_group)
usage_lay.setSpacing(8)
self.usage_total = StatCard()
self.usage_in = StatCard()
self.usage_out = StatCard()
self.usage_cache = StatCard()
self.usage_cost = StatCard()
self.usage_calls = StatCard()
for i, card in enumerate((self.usage_cost, self.usage_total,
self.usage_in, self.usage_out, self.usage_cache)):
usage_lay.addWidget(card, 0, i)
self.usage_calls.setVisible(False) # rides on the cost tile's label
self.budget_card = BudgetCard()
self.budget_card.apply_btn.setIcon(icon("check"))
self.budget_card.apply_btn.clicked.connect(self._apply_budget)
usage_lay.addWidget(self.budget_card, 0, 3)
for col in range(4):
usage_lay.setColumnStretch(col, 1)
root.addWidget(self.usage_group)
# ---- Recent Activity ----------------------------------------------------
def _build_activity_section(self) -> None:
self.activity_group = QGroupBox()
self.activity_group.setObjectName("monSection")
act_lay = QVBoxLayout(self.activity_group)
self.activity_lbl = QLabel()
self.activity_lbl.setWordWrap(True)
self.activity_lbl.setTextFormat(Qt.RichText)
act_lay.addWidget(self.activity_lbl)
# ---- Resource Usage -------------------------------------------------------
def _build_resource_section(self, root: QVBoxLayout) -> None:
self.resource_group = QGroupBox()
self.resource_group.setObjectName("monSection")
res_lay = QHBoxLayout(self.resource_group)
res_lay.setSpacing(6)
def _pair():
if not self._res_first:
sep = QLabel(chr(0xB7))
sep.setObjectName("hint")
res_lay.addWidget(sep)
self._res_first = False
lbl = QLabel()
lbl.setObjectName("hint")
val = QLabel()
res_lay.addWidget(lbl)
res_lay.addWidget(val)
return lbl, val
def _bar_row():
lbl, val = _pair()
bar = QProgressBar()
bar.setVisible(False)
return lbl, bar, val
self.cpu_lbl, self.cpu_bar, self.cpu_val = _bar_row()
self.mem_lbl, self.mem_bar, self.mem_val = _bar_row()
self.diskfree_lbl, self.diskfree_val = _pair()
self.disk_lbl, self.disk_val = QLabel(), QLabel()
self.network_lbl, self.network_val = QLabel(), QLabel()
res_lay.addStretch(1)
root.addWidget(self.resource_group)
# ---- Audit Log preview --------------------------------------------------
def _build_audit_section(self, root: QVBoxLayout, action_logs_tab_visible: bool) -> None:
self.audit_group = QGroupBox()
self.audit_group.setObjectName("monSection")
audit_lay = QVBoxLayout(self.audit_group)
self.audit_lbl = QLabel()
self.audit_lbl.setWordWrap(True)
self.audit_lbl.setTextFormat(Qt.RichText)
audit_lay.addWidget(self.audit_lbl)
self.view_all_btn = QPushButton()
self.view_all_btn.setFlat(True)
self.view_all_btn.clicked.connect(lambda: self._on_view_all_action_logs())
audit_lay.addWidget(self.view_all_btn, 0, Qt.AlignRight)
self.audit_group.setVisible(action_logs_tab_visible)
root.addWidget(self.audit_group)
# ---- budget --------------------------------------------------------------
def _apply_budget(self) -> None:
"""Persist the spin box's value as the new budget — starts a fresh
remaining-balance window (spend before now is no longer counted)."""
ccy = (self.ctx.config.data.get("usage") or {}).get("currency", "USD")
ut.set_budget(self.ctx.config, self.budget_card.budget_spin.value(), ccy)
self.ctx.save()
self._refresh_budget()
def _refresh_budget(self) -> None:
from ....core import model_pricing as mp
pricing = {**ut.DEFAULT_PRICING, **(self.ctx.config.data.get("usage") or {})}
status = ut.budget_status(self.ctx.config)
if status is None:
self.budget_card.set(tr("usage.budget_title"), "—", tr("usage.budget_no_budget"))
self.budget_card.budget_spin.setValue(0.0)
return
amount_disp = mp.convert(status["amount_usd"], "USD",
pricing.get("currency", "USD"), self.ctx.config)
value = (f"{ut.format_cost(status['remaining_usd'], pricing, digits=2)}"
f" / {ut.format_cost(status['amount_usd'], pricing, digits=2)}")
pct = int(round(status["pct_used"] * 100))
sub = tr("usage.budget_over_warning") if status["over_85"] else tr("usage.budget_used_pct", pct=pct)
self.budget_card.set(tr("usage.budget_title"), value, sub, warn=status["over_85"])
if not self.budget_card.budget_spin.hasFocus():
self.budget_card.budget_spin.setValue(round(amount_disp, 2))
# ---- resource usage --------------------------------------------------------
def _refresh_resource_usage(self) -> None:
try:
import psutil
except ImportError:
self._set_resource_na()
return
try:
own = psutil.Process()
own_cpu = own.cpu_percent(interval=None)
own_mem = own.memory_info().rss
except Exception:
own, own_cpu, own_mem = None, 0.0, 0
self.cpu_bar.setValue(int(min(own_cpu, 100)))
self.cpu_val.setText(f"{own_cpu:.0f}%")
try:
total_mem = psutil.virtual_memory().total
mem_pct = int(own_mem * 100 / total_mem) if total_mem else 0
except Exception:
mem_pct = 0
self.mem_bar.setValue(min(mem_pct, 100))
try:
self.mem_val.setText(f"{fmt_bytes(own_mem)}/{fmt_bytes(total_mem)}")
except Exception: # noqa: BLE001
self.mem_val.setText(fmt_bytes(own_mem))
try:
free = psutil.disk_usage(str(self.ctx.config.cowork_output_dir())).free
self.diskfree_val.setText(tr("monitoring.overview_disk_free", size=fmt_bytes(free)))
except Exception: # noqa: BLE001
self.diskfree_val.setText(tr("monitoring.na"))
now = time.monotonic()
try:
io = own.io_counters() if own is not None else None
disk_bytes = (io.read_bytes + io.write_bytes) if io is not None else None
except Exception:
disk_bytes = None
try:
net = psutil.net_io_counters()
net_bytes = net.bytes_sent + net.bytes_recv
except Exception:
net_bytes = None
prev = self._last_io_sample
self._last_io_sample = (now, disk_bytes, net_bytes)
na = tr("monitoring.na")
if prev and disk_bytes is not None and prev[1] is not None and now > prev[0]:
rate = max(0.0, (disk_bytes - prev[1]) / (now - prev[0]))
self.disk_val.setText(f"{fmt_bytes(rate)}/s")
else:
self.disk_val.setText(na)
if prev and net_bytes is not None and prev[2] is not None and now > prev[0]:
rate = max(0.0, (net_bytes - prev[2]) / (now - prev[0]))
self.network_val.setText(f"{fmt_bytes(rate)}/s")
else:
self.network_val.setText(na)
def _set_resource_na(self) -> None:
na = tr("monitoring.na")
self.cpu_bar.setValue(0)
self.cpu_val.setText(na)
self.mem_bar.setValue(0)
self.mem_val.setText(na)
self.disk_val.setText(na)
self.network_val.setText(na)
# ---- usage cards -----------------------------------------------------------
def _activity_line(self, event: dict) -> str:
ok = event.get("ok", True)
if ok:
mark = f"<span style='color:{DOT_GREEN};'>✓</span>"
elif event.get("kind") == "security_block":
mark = f"<span style='color:{DOT_AMBER};'>!</span>"
else:
mark = f"<span style='color:{DOT_RED};'>✗</span>"
name = event.get("name", "") or event.get("kind", "")
rel = relative_time(event.get("ts", ""))
muted = current_palette().text_muted
suffix = f" <span style='color:{muted};'>— {rel}</span>" if rel else ""
return f"{mark} {name}{suffix}"
def _refresh_usage_cards(self) -> None:
from ....core import model_pricing as mp
mp.sync_to_usage(self.ctx.config) # cost total comes straight from the price table
pricing = {**ut.DEFAULT_PRICING, **(self.ctx.config.data.get("usage") or {})}
events = ut.load_events()
s = ut.summarize(events)
costs = ut.cost_usd_events(events, pricing)
self.usage_total.set(tr("dashboard.card_total"), fmt_tokens(s["total"]))
self.usage_calls.set(tr("monitoring.overview_calls"), str(s["turns"]), "")
self.usage_in.set(tr("dashboard.card_in"), fmt_tokens(s["in"]),
ut.format_cost(costs["in"], pricing))
self.usage_out.set(tr("dashboard.card_out"), fmt_tokens(s["out"]),
ut.format_cost(costs["out"], pricing))
self.usage_cache.set(tr("dashboard.card_cache"), fmt_tokens(s["cache"]),
ut.format_cost(costs["cache"], pricing))
self.usage_cost.set(
f'{tr("dashboard.card_cost")} · {s["turns"]} {tr("monitoring.overview_calls")}',
ut.format_cost(sum(costs.values()), pricing, digits=2))
self._refresh_budget()
# ---- public API used by the container ---------------------------------
def refresh(self, events: List[dict]) -> None:
"""Full refresh — resource usage + usage cards + sandbox/permissions
+ recent activity + audit preview. ``events`` is the already-loaded
(local-or-shared) audit log, shared with the event-table tabs so the
decision of which source to read from is made exactly once per
refresh tick."""
self._refresh_resource_usage()
self._refresh_usage_cards()
self.sandbox_card.refresh()
recent = sorted(events, key=lambda e: e.get("ts", ""), reverse=True)
if recent:
self.activity_lbl.setText("<br>".join(self._activity_line(e) for e in recent[:6]))
self.audit_lbl.setText("<br>".join(
f"{e.get('ts', '')} — {e.get('name', '')}" for e in recent[:4]))
else:
self.activity_lbl.setText(tr("monitoring.overview_no_activity"))
self.audit_lbl.setText(tr("monitoring.overview_no_activity"))
def retranslate(self) -> None:
self.usage_group.setTitle(tr("monitoring.overview_usage_title").upper().replace("&", "&&"))
self.budget_card.apply_btn.setToolTip(tr("usage.budget_apply_tooltip"))
self.budget_card.budget_spin.setToolTip(tr("usage.budget_spin_tooltip"))
self.activity_group.setTitle(tr("monitoring.overview_activity_title").upper())
self.resource_group.setTitle(tr("monitoring.overview_resource_title").upper())
self.cpu_lbl.setText(tr("monitoring.overview_res_cpu"))
self.mem_lbl.setText(tr("monitoring.overview_res_mem"))
self.diskfree_lbl.setText(tr("monitoring.overview_disk_label"))
self.disk_lbl.setText(tr("monitoring.overview_res_disk"))
self.network_lbl.setText(tr("monitoring.overview_res_network"))
self.pricing_panel.retranslate()
self.sandbox_card.retranslate()
self.audit_group.setTitle(tr("monitoring.overview_audit_title").upper())
self.view_all_btn.setText(tr("monitoring.overview_view_all"))
@@ -0,0 +1,182 @@
"""Model Pricing panel — the editable price-table card on the Overview page
(currency picker, import/export/add/auto-link/delete, and the table itself).
Extracted from ``ui/monitoring_tab.py``'s pricing-table construction and
``_reload_pricing_table``/``_import_pricing``/``_export_pricing``/
``_add_pricing_row``/``_autolink_pricing``/``_delete_pricing_row``.
"""
from __future__ import annotations
from typing import Callable
from PySide6.QtWidgets import (
QComboBox, QGroupBox, QHBoxLayout, QHeaderView, QLabel, QPushButton,
QTableWidget, QTableWidgetItem, QVBoxLayout,
)
from ....core import usage_tracker as ut
from ....i18n import tr
from ....ui.icons import icon
class PricingPanel(QGroupBox):
def __init__(self, ctx, on_status_message: Callable[[str], None]):
super().__init__()
self.ctx = ctx
self._on_status_message = on_status_message
self._worker = None
self.setObjectName("monSection")
pg = QVBoxLayout(self)
phdr = QHBoxLayout()
self.ccy_lbl = QLabel()
self.ccy_lbl.setObjectName("hint")
self.ccy = QComboBox()
for cur in ut.SUPPORTED_CURRENCIES:
self.ccy.addItem(cur, cur)
pidx = self.ccy.findData((self.ctx.config.data.get("usage") or {}).get("currency", "USD"))
self.ccy.setCurrentIndex(max(0, pidx))
self.ccy.currentIndexChanged.connect(self._reload_table)
phdr.addWidget(self.ccy_lbl)
phdr.addWidget(self.ccy)
phdr.addStretch(1)
self.import_btn = QPushButton()
self.import_btn.setIcon(icon("download"))
self.import_btn.clicked.connect(self._import_pricing)
self.export_btn = QPushButton()
self.export_btn.setIcon(icon("upload"))
self.export_btn.clicked.connect(self._export_pricing)
self.add_btn = QPushButton()
self.add_btn.setIcon(icon("plus"))
self.add_btn.clicked.connect(self._add_pricing_row)
self.link_btn = QPushButton()
self.link_btn.setIcon(icon("refresh"))
self.link_btn.clicked.connect(self._autolink_pricing)
self.del_btn = QPushButton()
self.del_btn.setIcon(icon("trash"))
self.del_btn.clicked.connect(self._delete_pricing_row)
for b in (self.import_btn, self.export_btn, self.add_btn, self.link_btn, self.del_btn):
phdr.addWidget(b)
pg.addLayout(phdr)
self.table = QTableWidget(0, 5)
self.table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.table.verticalHeader().setVisible(False)
self.table.setEditTriggers(QTableWidget.NoEditTriggers)
self.table.setSelectionBehavior(QTableWidget.SelectRows)
pg.addWidget(self.table, 1)
self._reload_table()
def retranslate(self) -> None:
self.setTitle(tr("monitoring.pricing_title").upper())
self.ccy_lbl.setText(tr("monitoring.pricing_currency"))
self.import_btn.setText(tr("monitoring.pricing_import"))
self.export_btn.setText(tr("monitoring.pricing_export"))
self.add_btn.setText(tr("monitoring.pricing_add"))
self.link_btn.setText(tr("monitoring.pricing_autolink"))
self.del_btn.setText(tr("monitoring.pricing_delete"))
self.table.setHorizontalHeaderLabels([
tr("monitoring.pricing_col_model"), tr("monitoring.pricing_col_context"),
tr("monitoring.pricing_col_maxout"), tr("monitoring.pricing_col_input"),
tr("monitoring.pricing_col_output")])
def _reload_table(self, *_a) -> None:
from ....core import model_pricing as mp
to_ccy = self.ccy.currentData() or "USD"
entries = mp.list_entries(self.ctx.config)
self.table.setRowCount(len(entries))
for r, e in enumerate(entries):
in_v = mp.convert(e.get("input_price", 0), e.get("input_ccy", "USD"), to_ccy, self.ctx.config)
out_v = mp.convert(e.get("output_price", 0), e.get("output_ccy", "USD"), to_ccy, self.ctx.config)
vals = [e.get("model", ""), e.get("context_length", ""), e.get("max_output", ""),
f"{mp.format_price(in_v, to_ccy)} / {e.get('input_unit', '')}",
f"{mp.format_price(out_v, to_ccy)} / {e.get('output_unit', '')}"]
for c, v in enumerate(vals):
self.table.setItem(r, c, QTableWidgetItem(str(v)))
def _import_pricing(self) -> None:
from PySide6.QtWidgets import QFileDialog, QMessageBox
from ....core import model_pricing as mp
path, _ = QFileDialog.getOpenFileName(
self, tr("monitoring.pricing_import"), "", "Pricing (*.xlsx *.csv)")
if not path:
return
default_ccy = self.ccy.currentData() or "USD"
try:
imported = mp.import_table(path, default_ccy=default_ccy)
except ValueError as exc:
QMessageBox.warning(self, tr("monitoring.pricing_title"), str(exc))
return
merged = {e["model"]: e for e in mp.list_entries(self.ctx.config)}
for e in imported:
merged[e["model"]] = e
mp.save_entries(self.ctx.config, list(merged.values()))
self.ctx.save()
self._reload_table()
self._on_status_message(tr("monitoring.pricing_imported", n=len(imported)))
def _export_pricing(self) -> None:
from PySide6.QtWidgets import QFileDialog
from ....core import model_pricing as mp
path, _ = QFileDialog.getSaveFileName(
self, tr("monitoring.pricing_export"), "model_pricing_template.xlsx", "Excel (*.xlsx)")
if not path:
return
mp.export_template(path)
self._on_status_message(tr("monitoring.pricing_exported"))
def _add_pricing_row(self) -> None:
from PySide6.QtWidgets import QInputDialog
from ....core import model_pricing as mp
name, ok = QInputDialog.getText(self, tr("monitoring.pricing_add"),
tr("monitoring.pricing_add_prompt"))
name = (name or "").strip()
if not ok or not name:
return
ccy = self.ccy.currentData() or "USD"
mp.add_entry(self.ctx.config, mp.entry_from_row(
[name, "", "", "0", "Million tokens", "0", "Million tokens"], default_ccy=ccy))
self.ctx.save()
self._reload_table()
def _autolink_pricing(self) -> None:
from ....core import model_pricing as mp
from ....core.worker import AgentWorker
if self._worker is not None:
return
self.link_btn.setEnabled(False)
ctx = self.ctx
ccy = self.ccy.currentData() or "USD"
def job(_w):
return {"entries": mp.auto_link(ctx, default_ccy=ccy)}
def done(r):
self._worker = None
self.link_btn.setEnabled(True)
self.ctx.save()
self._reload_table()
self._on_status_message(tr("monitoring.pricing_linked", n=len(r.get("entries", []))))
def failed(_e):
self._worker = None
self.link_btn.setEnabled(True)
w = AgentWorker(job)
w.finished_ok.connect(done)
w.failed.connect(failed)
self._worker = w
w.start()
def _delete_pricing_row(self) -> None:
from ....core import model_pricing as mp
row = self.table.currentRow()
entries = mp.list_entries(self.ctx.config)
if 0 <= row < len(entries):
del entries[row]
mp.save_entries(self.ctx.config, entries)
self.ctx.save()
self._reload_table()
+135
View File
@@ -0,0 +1,135 @@
"""Sandbox Details card — the current sandbox id/status/uptime/resource
limits/network state, with a collapsible fold that ALSO nests the
Permissions card inside it (see ``security_settings_tab.PermissionsCard``),
exactly matching the pre-refactor ``ui/monitoring_tab.py`` layout: Sandbox
and Permissions answer the same question ("what is the agent allowed to
touch?"), so they share one fold rather than being two independent
top-level sections. This card is embedded inside ``overview_tab.OverviewTab``
at the same position the original ``QGroupBox`` occupied — no new top-level
tab is added, so the visible UI is unchanged.
"""
from __future__ import annotations
import os
import time
from datetime import datetime
from typing import Callable
from PySide6.QtCore import Qt
from PySide6.QtWidgets import QGroupBox, QHBoxLayout, QLabel, QPushButton, QVBoxLayout, QWidget
from ....i18n import tr
from ..shared.badges import apply_badge
from ..shared.layout_helpers import kv_row
from ..shared.open_settings import open_settings_and_notify
from .security_settings_tab import PermissionsCard
class SandboxDetailsCard(QGroupBox):
def __init__(self, ctx, on_settings_changed: Callable[[], None]):
super().__init__()
self._ctx = ctx
self._on_settings_changed = on_settings_changed
self.setObjectName("monSection")
sbx_lay = QVBoxLayout(self)
self.summary_lbl = QLabel()
self.summary_lbl.setWordWrap(True)
sbx_lay.addWidget(self.summary_lbl)
self.more_btn = QPushButton()
self.more_btn.setObjectName("co4eSectionAction")
self.more_btn.setFlat(True)
self.more_btn.setCheckable(True)
self.more_btn.setCursor(Qt.PointingHandCursor)
sbx_lay.addWidget(self.more_btn, 0, Qt.AlignLeft)
self._detail = QWidget()
self._detail.setVisible(False)
self.more_btn.toggled.connect(self._detail.setVisible)
self.more_btn.toggled.connect(self._sync_more_label)
sbx_lay.addWidget(self._detail)
detail_lay = QVBoxLayout(self._detail)
detail_lay.setContentsMargins(0, 4, 0, 0)
self.id_lbl, self.id_val = kv_row(detail_lay)
self.status_lbl, self.status_val = kv_row(detail_lay)
self.status_val.setObjectName("badgeSuccess")
self.created_lbl, self.created_val = kv_row(detail_lay)
self.uptime_lbl, self.uptime_val = kv_row(detail_lay)
limits_row = QHBoxLayout()
self.limits_lbl = QLabel()
self.limits_lbl.setObjectName("hint")
self.limits_lbl.setWordWrap(True)
self.edit_btn = QPushButton()
self.edit_btn.setFlat(True)
self.edit_btn.clicked.connect(self._open_settings)
limits_row.addWidget(self.limits_lbl, 1)
limits_row.addWidget(self.edit_btn)
detail_lay.addLayout(limits_row)
self.net_lbl, self.net_val = kv_row(detail_lay)
# The Permissions card is nested inside THIS fold, not a sibling
# section — matches the original layout exactly.
self.permissions_card = PermissionsCard(ctx, on_settings_changed)
detail_lay.addWidget(self.permissions_card)
def _open_settings(self) -> None:
open_settings_and_notify(self._ctx, self, self._on_settings_changed)
def _sync_more_label(self, *_a) -> None:
"""Label the fold with what it will do next."""
open_ = self.more_btn.isChecked()
self.more_btn.setText(("▾ " if open_ else "▸ ") + tr("monitoring.overview_sbx_detail"))
def retranslate(self) -> None:
self.setTitle(tr("monitoring.overview_sandbox_details_title").upper().replace("&", "&&"))
self.id_lbl.setText(tr("monitoring.overview_sandbox_id"))
self.status_lbl.setText(tr("monitoring.overview_status"))
self.created_lbl.setText(tr("monitoring.overview_created"))
self.uptime_lbl.setText(tr("monitoring.overview_uptime"))
self.edit_btn.setText(tr("monitoring.overview_edit"))
self.net_lbl.setText(tr("monitoring.overview_network_label"))
self.permissions_card.retranslate()
self._sync_more_label()
def refresh(self) -> None:
sec = self._ctx.config.agent_security
net_blocked = bool(sec.get("block_network"))
self.id_val.setText(f"sbx_{os.getpid():x}")
self.status_val.setText(tr("monitoring.overview_status_running"))
self.created_val.setText(datetime.fromtimestamp(self._ctx.started_at).strftime("%H:%M:%S"))
uptime_s = max(0, int(time.time() - self._ctx.started_at))
h, rem = divmod(uptime_s, 3600)
m, s = divmod(rem, 60)
self.uptime_val.setText(f"{h}h {m}m {s}s" if h else f"{m}m {s}s")
limit_parts = []
if sec.get("resource_limit_cpu_percent"):
limit_parts.append(f"CPU {sec['resource_limit_cpu_percent']}%")
if sec.get("resource_limit_memory_mb"):
limit_parts.append(f"MEM {sec['resource_limit_memory_mb']}MB")
if sec.get("resource_limit_disk_mb"):
limit_parts.append(f"DISK {sec['resource_limit_disk_mb']}MB")
limits_text = ", ".join(limit_parts) if limit_parts else tr("monitoring.na")
self.limits_lbl.setText(tr("monitoring.overview_resource_limits") + ": " + limits_text)
self.net_val.setText(
tr("monitoring.overview_network_disabled") if net_blocked
else tr("monitoring.overview_network_enabled"))
apply_badge(self.net_val, "badgeWarn" if net_blocked else "badgeSuccess")
# The one line the wireframe shows; the detail above stays a fold away.
self.summary_lbl.setText(" · ".join([
f'{tr("monitoring.overview_perm_fs")}: {tr("monitoring.overview_perm_fs_value")}',
f'{tr("monitoring.overview_perm_network")}: '
f'{tr("monitoring.overview_network_disabled") if net_blocked else tr("monitoring.overview_network_enabled")}',
f'{tr("monitoring.overview_perm_process")}: {tr("monitoring.overview_perm_process_value")}',
f'{tr("monitoring.overview_resource_limits")}: {limits_text}',
]))
self._sync_more_label()
self.permissions_card.refresh()
@@ -0,0 +1,47 @@
"""Security Events tab — the audit log filtered to ``kind="security_block"``.
Extracted from ``ui/monitoring_tab.py``'s security-events wiring inside
``MonitoringTab.__init__``.
"""
from __future__ import annotations
from typing import Callable, List
from PySide6.QtWidgets import QLineEdit, QPushButton, QWidget
from ....i18n import tr
from ..shared.ai_filter import start_ai_filter
from ..shared.event_table import EventTable
from ..shared.filter_scaffold import build_filter_scaffold
class SecurityEventsTab(QWidget):
def __init__(self, ctx, on_refresh_all: Callable[[], None]):
super().__init__()
self._ctx = ctx
self._ai_state: dict = {}
# Security events are always ok=False, so this table trades the
# tick/cross column for a tinted Action column (see EventTable).
self.table = EventTable(show_result=False)
parts = build_filter_scaffold(
self, self.table, on_refresh=on_refresh_all,
title_key="monitoring.security_events_title",
with_search=True, with_detail=True, on_ai_filter=self._start_ai_filter)
self.title_lbl = parts["title_lbl"]
self.title_key = parts["title_key"]
self.title_refresh_btn = parts["title_refresh_btn"]
self.filter_edit = parts["filter_edit"]
self.ai_filter_btn = parts["ai_filter_btn"]
self.detail_panel = parts["detail_panel"]
def set_events(self, events: List[dict]) -> None:
self.table.set_events(events)
def retranslate(self) -> None:
self.table.retranslate()
self.filter_edit.setPlaceholderText(tr("monitoring.filter_placeholder"))
self.detail_panel.retranslate()
self.title_lbl.setText(tr(self.title_key))
self.title_refresh_btn.setText(tr("monitoring.refresh"))
def _start_ai_filter(self, search: QLineEdit, ai_btn: QPushButton) -> None:
start_ai_filter(self._ctx, search, ai_btn, self._ai_state)
@@ -0,0 +1,59 @@
"""Permissions card — "what is the agent allowed to touch?", displayed
nested inside the Sandbox Details card's expandable fold (see
``sandbox_tab.SandboxDetailsCard``), exactly as in the pre-refactor
``ui/monitoring_tab.py`` (``self._sbx_detail.layout().addWidget(self.ov_permissions_group)``).
Editing still opens the same Settings dialog as the Sandbox card's own
"Edit" button — this card only DISPLAYS ``ctx.config.agent_security``, it
does not host its own settings-editing UI.
"""
from __future__ import annotations
from typing import Callable
from PySide6.QtCore import Qt
from PySide6.QtWidgets import QGroupBox, QPushButton, QVBoxLayout
from ....i18n import tr
from ..shared.badges import apply_badge
from ..shared.layout_helpers import kv_row
from ..shared.open_settings import open_settings_and_notify
class PermissionsCard(QGroupBox):
def __init__(self, ctx, on_settings_changed: Callable[[], None]):
super().__init__()
self._ctx = ctx
self._on_settings_changed = on_settings_changed
self.setObjectName("monSection")
perm_lay = QVBoxLayout(self)
self.fs_lbl, self.fs_val = kv_row(perm_lay)
self.network_lbl, self.network_val = kv_row(perm_lay)
self.process_lbl, self.process_val = kv_row(perm_lay)
self.env_lbl, self.env_val = kv_row(perm_lay)
self.edit_btn = QPushButton()
self.edit_btn.setFlat(True)
self.edit_btn.clicked.connect(self._open_settings)
perm_lay.addWidget(self.edit_btn, 0, Qt.AlignLeft)
def _open_settings(self) -> None:
open_settings_and_notify(self._ctx, self, self._on_settings_changed)
def retranslate(self) -> None:
self.setTitle(tr("monitoring.overview_permissions_title").upper())
self.fs_lbl.setText(tr("monitoring.overview_perm_fs"))
self.fs_val.setText(tr("monitoring.overview_perm_fs_value"))
self.network_lbl.setText(tr("monitoring.overview_perm_network"))
self.process_lbl.setText(tr("monitoring.overview_perm_process"))
self.process_val.setText(tr("monitoring.overview_perm_process_value"))
self.env_lbl.setText(tr("monitoring.overview_perm_env"))
self.env_val.setText(tr("monitoring.overview_perm_env_value"))
self.edit_btn.setText(tr("monitoring.overview_edit"))
def refresh(self) -> None:
net_blocked = bool(self._ctx.config.agent_security.get("block_network"))
self.network_val.setText(
tr("monitoring.overview_perm_network_blocked") if net_blocked
else tr("monitoring.overview_perm_network_allowed"))
apply_badge(self.network_val, "badgeWarn" if net_blocked else "badgeSuccess")
+75
View File
@@ -0,0 +1,75 @@
"""Task 4b — agent_security <-> agent_security_alert circular dependency is gone.
Before this fix, ``agent_security_alert.py`` imported ``SecurityVerdict`` from
``agent_security.py`` at module level (for the ``notify_admin`` type
annotation), while ``agent_security.py`` deferred-imported
``agent_security_alert.notify_admin`` inside ``enforce_prompt``/
``enforce_command`` — an architectural cycle only avoided at runtime by
pushing that second import inside a function body.
``SecurityVerdict``/``SecurityBlocked`` now live in the dependency-free leaf
module ``agent_security_types.py``. ``agent_security_alert.py`` imports the
type from there instead of from ``agent_security.py``, which lets
``agent_security.py`` import ``agent_security_alert.notify_admin`` at module
top level with no cycle.
"""
from __future__ import annotations
from cowork_local.core import (
agent_security,
agent_security_alert,
agent_security_types,
)
def test_shared_types_live_in_the_leaf_module() -> None:
assert agent_security.SecurityVerdict is agent_security_types.SecurityVerdict
assert agent_security.SecurityBlocked is agent_security_types.SecurityBlocked
assert agent_security_alert.SecurityVerdict is agent_security_types.SecurityVerdict
def test_agent_security_alert_no_longer_imports_agent_security() -> None:
assert "agent_security" not in agent_security_alert.__dict__
def test_notify_admin_imported_at_module_top_level_in_agent_security() -> None:
assert agent_security.notify_admin is agent_security_alert.notify_admin
def test_enforce_command_still_blocks_and_alerts_like_before(monkeypatch) -> None:
class _FakeProvider:
def chat(self, messages, tools=None):
return {"content": '{"allowed": false, "reason": "destructive"}'}
class _Config:
data = {"agent_security": {"enabled": True, "validate_commands": True,
"command_ai_check": True}}
ms365 = {}
@property
def agent_security(self):
return self.data["agent_security"]
notify_calls = []
record_calls = []
monkeypatch.setattr(agent_security, "notify_admin",
lambda config, verdict, detail="": notify_calls.append((verdict, detail)))
from cowork_local.core import audit_log
monkeypatch.setattr(audit_log, "record",
lambda *a, **k: record_calls.append((a, k)))
emitted = []
raised = False
try:
agent_security.enforce_command(
_FakeProvider(), "run_command", {"command": "rm -rf /"}, _Config(),
emit=emitted.append,
)
except agent_security.SecurityBlocked as exc:
raised = True
assert exc.verdict.layer == "command"
assert raised is True
assert notify_calls
assert record_calls
assert emitted and emitted[0]["type"] == "notice"
+95
View File
@@ -0,0 +1,95 @@
"""Task 2 — CanonicalAuditLogger.
Verifies: (1) the infrastructure class itself round-trips events correctly
and mirrors to a shared dir, (2) ``core/audit_log.py``'s wrapper functions
still behave exactly as before (same signatures, same dict schema, same
never-raise guarantee), and (3) old-format raw dicts (as written by the
pre-refactor ``core/audit_log.py``) still load correctly for backward
compatibility.
"""
from __future__ import annotations
import json
from cowork_local.core import audit_log
from cowork_local.infrastructure.telemetry.audit_logger import (
KIND_MCP_CALL,
KIND_SECURITY_BLOCK,
CanonicalAuditEvent,
CanonicalAuditLogger,
)
def test_record_and_load_round_trip(tmp_path) -> None:
logger = CanonicalAuditLogger(tmp_path)
logger.set_identity("alice", "machine-1", role="admin")
logger.record(KIND_SECURITY_BLOCK, "run_command", False, detail="blocked it")
events = logger.load_events()
assert len(events) == 1
e = events[0]
assert e.kind == KIND_SECURITY_BLOCK
assert e.name == "run_command"
assert e.ok is False
assert e.detail == "blocked it"
assert e.account == "alice"
assert e.machine == "machine-1"
assert e.role == "admin"
def test_load_events_filters_by_kind(tmp_path) -> None:
logger = CanonicalAuditLogger(tmp_path)
logger.record(KIND_SECURITY_BLOCK, "a", False)
logger.record(KIND_MCP_CALL, "b", True)
only_mcp = logger.load_events(kind=KIND_MCP_CALL)
assert [e.name for e in only_mcp] == ["b"]
def test_shared_dir_mirroring(tmp_path) -> None:
shared = tmp_path / "shared"
logger = CanonicalAuditLogger(tmp_path / "audit")
logger.set_identity("bob", "machine-2", shared_dir=str(shared))
logger.record(KIND_MCP_CALL, "tool_x", True)
mirrored_files = list((shared / "telemetry" / "audit").glob("machine-2-*.jsonl"))
assert len(mirrored_files) == 1
def test_from_dict_is_tolerant_of_old_partial_rows() -> None:
old_row = {"ts": "2024-01-01T00:00:00", "kind": "tool_call", "name": "x", "ok": True}
event = CanonicalAuditEvent.from_dict(old_row)
assert event.detail == ""
assert event.account == ""
def test_record_never_raises_on_bad_directory(tmp_path) -> None:
bad_dir = tmp_path / "some_file.txt"
bad_dir.write_text("not a directory")
logger = CanonicalAuditLogger(bad_dir / "audit")
logger.record(KIND_SECURITY_BLOCK, "x", False) # must not raise
def test_core_audit_log_wrapper_same_schema_as_before(tmp_path, monkeypatch) -> None:
monkeypatch.setattr(audit_log, "AUDIT_DIR", tmp_path)
monkeypatch.setattr(audit_log, "_logger",
audit_log.CanonicalAuditLogger(tmp_path))
audit_log.set_identity("carol", "machine-3", role="user")
audit_log.record("permission", "install_package", True, detail="ok", agent_role="cowork")
events = audit_log.load_events()
assert len(events) == 1
e = events[0]
assert set(e.keys()) == {"ts", "kind", "agent_role", "name", "ok", "detail",
"account", "role", "machine"}
assert e["kind"] == "permission"
assert e["name"] == "install_package"
assert e["ok"] is True
assert e["agent_role"] == "cowork"
assert e["account"] == "carol"
# Raw file on disk still uses the exact pre-refactor schema/keys.
raw_line = next((tmp_path).glob("*.jsonl")).read_text(encoding="utf-8").splitlines()[0]
raw = json.loads(raw_line)
assert list(raw.keys()) == ["ts", "kind", "agent_role", "name", "ok", "detail",
"account", "role", "machine"]
+60
View File
@@ -0,0 +1,60 @@
"""Task 4a — model_pricing <-> usage_tracker circular dependency is gone.
Before this fix, ``model_pricing.turn_cost_usd`` deferred-imported
``usage_tracker`` for its flat fallback rates, while ``usage_tracker.set_budget``
deferred-imported ``model_pricing`` for currency conversion — a real
architectural cycle, only avoided at runtime by pushing both imports inside
function bodies. Now ``model_pricing`` is a leaf module (it owns its own
fallback rates) and ``usage_tracker`` imports it at module top level.
"""
from __future__ import annotations
import copy
import sys
from cowork_local.config import AppConfig, DEFAULT_CONFIG
from cowork_local.core import model_pricing, usage_tracker
def test_model_pricing_does_not_depend_on_usage_tracker_module_level() -> None:
assert "usage_tracker" not in model_pricing.__dict__
assert "usage_tracker" not in getattr(model_pricing, "__all__", [])
def test_usage_tracker_imports_model_pricing_at_top_level() -> None:
assert usage_tracker.mp is model_pricing
def test_turn_cost_usd_fallback_matches_pre_refactor_default_rates(tmp_path) -> None:
config = AppConfig(data=copy.deepcopy(DEFAULT_CONFIG), path=tmp_path / "config.json")
# No matching row in the price table and no override in config.data["usage"]
# -> falls back to the flat rates that used to live in
# usage_tracker.DEFAULT_PRICING (0.5 in / 1.5 out USD per 1M tokens).
cost = model_pricing.turn_cost_usd("some-unknown-model", 1_000_000, 1_000_000, config)
assert cost == 0.5 + 1.5
def test_turn_cost_usd_honours_usage_override_like_before(tmp_path) -> None:
config = AppConfig(data=copy.deepcopy(DEFAULT_CONFIG), path=tmp_path / "config.json")
config.data.setdefault("usage", {})["price_per_mtok_in_usd"] = 2.0
config.data["usage"]["price_per_mtok_out_usd"] = 4.0
cost = model_pricing.turn_cost_usd("some-unknown-model", 1_000_000, 1_000_000, config)
assert cost == 2.0 + 4.0
def test_set_budget_still_converts_via_model_pricing(tmp_path) -> None:
config = AppConfig(data=copy.deepcopy(DEFAULT_CONFIG), path=tmp_path / "config.json")
usage_tracker.set_budget(config, 100.0, currency="USD")
status = usage_tracker.budget_status(config)
assert status is not None
assert status["amount_usd"] == 100.0
def test_no_import_time_cycle_when_loaded_fresh() -> None:
for name in ("cowork_local.core.model_pricing", "cowork_local.core.usage_tracker"):
sys.modules.pop(name, None)
import importlib
mp = importlib.import_module("cowork_local.core.model_pricing")
ut = importlib.import_module("cowork_local.core.usage_tracker")
assert ut.mp is mp
+56
View File
@@ -0,0 +1,56 @@
"""Task 1 (sub-step 6d) — AgentStatusTab widget smoke test."""
from __future__ import annotations
import copy
import os
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
QApplication = pytest.importorskip("PySide6.QtWidgets").QApplication
from cowork_local.config import AppConfig, DEFAULT_CONFIG
from cowork_local.presentation.monitoring.tabs.agent_status_tab import AgentStatusTab
@pytest.fixture(scope="module")
def qapp():
app = QApplication.instance() or QApplication([])
yield app
class _FakeCowork:
def active_workers(self):
return [1, 2]
class _FakeTaskScheduler:
def running_count(self):
return 3
class _FakeCtx:
def __init__(self, tmp_path):
self.config = AppConfig(data=copy.deepcopy(DEFAULT_CONFIG), path=tmp_path / "c.json")
self.config.data["agent_security"]["enabled"] = True
def test_agent_status_tab_has_no_search_or_detail(qapp, tmp_path) -> None:
tab = AgentStatusTab(_FakeCtx(tmp_path), on_refresh_all=lambda: None)
assert not hasattr(tab, "filter_edit")
assert not hasattr(tab, "detail_panel")
def test_refresh_populates_six_rows_with_live_counts(qapp, tmp_path) -> None:
tab = AgentStatusTab(_FakeCtx(tmp_path), on_refresh_all=lambda: None,
cowork=_FakeCowork(), task_scheduler=_FakeTaskScheduler())
tab.refresh()
assert tab.table.rowCount() == 6
assert tab.table.item(0, 0).text() # Cowork row has a label
assert tab.table.cellWidget(0, 1) is not None # badge pill widget
def test_retranslate_does_not_raise(qapp, tmp_path) -> None:
tab = AgentStatusTab(_FakeCtx(tmp_path), on_refresh_all=lambda: None)
tab.retranslate()
+59
View File
@@ -0,0 +1,59 @@
"""Task 1 (sub-step 6c) — SecurityEventsTab / McpTab / ActionLogsTab.
Widget smoke tests against the offscreen QPA platform (see
test_monitoring_event_widgets.py for why no pytest-qt is needed). Verifies
each tab wires build_filter_scaffold correctly, exposes the attributes the
container's retranslate loop needs, and forwards set_events to its table.
"""
from __future__ import annotations
import os
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
QApplication = pytest.importorskip("PySide6.QtWidgets").QApplication
from cowork_local.presentation.monitoring.tabs.action_logs_tab import ActionLogsTab
from cowork_local.presentation.monitoring.tabs.mcp_tab import McpTab
from cowork_local.presentation.monitoring.tabs.security_events_tab import SecurityEventsTab
@pytest.fixture(scope="module")
def qapp():
app = QApplication.instance() or QApplication([])
yield app
@pytest.mark.parametrize("cls,expected_title_key", [
(SecurityEventsTab, "monitoring.security_events_title"),
(McpTab, "monitoring.mcp_history_title"),
(ActionLogsTab, "monitoring.action_logs_title"),
])
def test_tab_exposes_container_facing_attributes(qapp, cls, expected_title_key) -> None:
refresh_calls = []
tab = cls(ctx=None, on_refresh_all=lambda: refresh_calls.append(1))
assert tab.title_key == expected_title_key
assert tab.filter_edit is not None
assert tab.detail_panel is not None
tab.title_refresh_btn.click()
assert refresh_calls == [1]
def test_set_events_forwards_to_table(qapp) -> None:
tab = SecurityEventsTab(ctx=None, on_refresh_all=lambda: None)
tab.set_events([{"ts": "2026-01-01T00:00:00", "kind": "security_block", "name": "x",
"ok": False, "detail": "d", "account": "a", "machine": "m"}])
assert tab.table.rowCount() == 1
def test_retranslate_does_not_raise(qapp) -> None:
tab = McpTab(ctx=None, on_refresh_all=lambda: None)
tab.retranslate() # must not raise
def test_ai_filter_noop_when_search_box_empty(qapp) -> None:
tab = ActionLogsTab(ctx=None, on_refresh_all=lambda: None)
tab.ai_filter_btn.click() # empty search text -> start_ai_filter no-ops, must not raise
+79
View File
@@ -0,0 +1,79 @@
"""Task 1 (sub-step 6b) — EventTable / EventDetailPanel widget smoke tests.
These instantiate real PySide6 widgets against the offscreen QPA platform
(no display needed, no pytest-qt dependency — a plain QApplication instance
is enough to construct/query widgets, only an actual event loop would need
more). Verifies the extraction into presentation/monitoring/shared/ wires up
without error and preserves the pre-refactor row/column behaviour.
"""
from __future__ import annotations
import os
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
QApplication = pytest.importorskip("PySide6.QtWidgets").QApplication
from cowork_local.presentation.monitoring.shared.event_detail_panel import EventDetailPanel
from cowork_local.presentation.monitoring.shared.event_table import EventTable
@pytest.fixture(scope="module")
def qapp():
app = QApplication.instance() or QApplication([])
yield app
def _sample_event(**overrides):
ev = {"ts": "2026-05-25T15:03:00", "kind": "security_block", "name": "run_command",
"ok": False, "detail": "blocked it", "account": "alice", "machine": "m1",
"agent_role": "cowork"}
ev.update(overrides)
return ev
def test_event_table_security_events_hides_result_column(qapp) -> None:
table = EventTable(show_result=False)
table.retranslate()
assert table.columnCount() == 6
def test_event_table_generic_shows_result_column(qapp) -> None:
table = EventTable(show_result=True)
table.retranslate()
assert table.columnCount() == 7
def test_set_events_populates_rows_newest_first(qapp) -> None:
table = EventTable(show_result=True)
table.set_events([
_sample_event(ts="2026-05-25T10:00:00", name="first"),
_sample_event(ts="2026-05-25T12:00:00", name="second"),
])
assert table.rowCount() == 2
assert table.item(0, 4).text() == "second"
assert table.item(1, 4).text() == "first"
def test_event_at_row_round_trips_full_event(qapp) -> None:
table = EventTable(show_result=False)
ev = _sample_event()
table.set_events([ev])
assert table.event_at_row(0) == ev
def test_apply_filter_hides_non_matching_rows(qapp) -> None:
table = EventTable(show_result=True)
table.set_events([_sample_event(name="run_command"), _sample_event(name="fetch_url")])
table.apply_filter("fetch")
hidden = [table.isRowHidden(r) for r in range(table.rowCount())]
assert hidden.count(True) == 1
def test_detail_panel_shows_event_without_error(qapp) -> None:
panel = EventDetailPanel()
panel.retranslate()
panel.show_event(_sample_event(), 0)
assert panel._detail_text == "blocked it"
+80
View File
@@ -0,0 +1,80 @@
"""Task 1 (sub-step 6f) — OverviewTab widget smoke test."""
from __future__ import annotations
import copy
import os
import time
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
QApplication = pytest.importorskip("PySide6.QtWidgets").QApplication
from cowork_local.config import AppConfig, DEFAULT_CONFIG
from cowork_local.presentation.monitoring.tabs.overview_tab import OverviewTab
@pytest.fixture(scope="module")
def qapp():
app = QApplication.instance() or QApplication([])
yield app
class _FakeCtx:
def __init__(self, tmp_path):
self.config = AppConfig(data=copy.deepcopy(DEFAULT_CONFIG), path=tmp_path / "c.json")
self.started_at = time.time() - 30
def save(self):
pass
def cowork_output_dir(self):
return "."
def _make_tab(tmp_path):
return OverviewTab(
_FakeCtx(tmp_path), on_status_message=lambda _m: None,
on_settings_changed=lambda: None, on_view_all_action_logs=lambda: None,
action_logs_tab_visible=True)
def test_construction_and_retranslate(qapp, tmp_path) -> None:
tab = _make_tab(tmp_path)
tab.retranslate() # must not raise
def test_refresh_with_no_events_shows_no_activity_text(qapp, tmp_path) -> None:
tab = _make_tab(tmp_path)
tab.retranslate()
tab.refresh(events=[])
assert tab.activity_lbl.text() != ""
def test_refresh_with_events_renders_activity_lines(qapp, tmp_path) -> None:
tab = _make_tab(tmp_path)
tab.retranslate()
tab.refresh(events=[
{"ts": "2026-01-01T00:00:00", "kind": "tool_call", "name": "run_command", "ok": True},
{"ts": "2026-01-01T00:00:05", "kind": "security_block", "name": "blocked", "ok": False},
])
assert "run_command" in tab.activity_lbl.text() or "blocked" in tab.activity_lbl.text()
def test_view_all_button_invokes_callback(qapp, tmp_path) -> None:
calls = []
tab = OverviewTab(
_FakeCtx(tmp_path), on_status_message=lambda _m: None,
on_settings_changed=lambda: None, on_view_all_action_logs=lambda: calls.append(1),
action_logs_tab_visible=True)
tab.view_all_btn.click()
assert calls == [1]
def test_audit_group_visibility_follows_constructor_flag(qapp, tmp_path) -> None:
hidden_tab = OverviewTab(
_FakeCtx(tmp_path), on_status_message=lambda _m: None,
on_settings_changed=lambda: None, on_view_all_action_logs=lambda: None,
action_logs_tab_visible=False)
assert hidden_tab.audit_group.isHidden() is True
+49
View File
@@ -0,0 +1,49 @@
"""Task 1 (sub-step 6f, follow-up) — PricingPanel, split out of overview_tab.py
to keep that file under the 400-line quality rule."""
from __future__ import annotations
import copy
import os
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
QApplication = pytest.importorskip("PySide6.QtWidgets").QApplication
from cowork_local.config import AppConfig, DEFAULT_CONFIG
from cowork_local.presentation.monitoring.tabs.pricing_panel import PricingPanel
@pytest.fixture(scope="module")
def qapp():
app = QApplication.instance() or QApplication([])
yield app
class _FakeCtx:
def __init__(self, tmp_path):
self.config = AppConfig(data=copy.deepcopy(DEFAULT_CONFIG), path=tmp_path / "c.json")
def save(self):
pass
def test_construction_and_retranslate(qapp, tmp_path) -> None:
panel = PricingPanel(_FakeCtx(tmp_path), on_status_message=lambda _m: None)
panel.retranslate()
def test_add_and_delete_pricing_row_round_trip(qapp, tmp_path, monkeypatch) -> None:
from PySide6.QtWidgets import QInputDialog
ctx = _FakeCtx(tmp_path)
panel = PricingPanel(ctx, on_status_message=lambda _m: None)
monkeypatch.setattr(QInputDialog, "getText", staticmethod(lambda *a, **k: ("gpt-test", True)))
panel._add_pricing_row()
assert panel.table.rowCount() == 1
assert panel.table.item(0, 0).text() == "gpt-test"
panel.table.selectRow(0)
panel._delete_pricing_row()
assert panel.table.rowCount() == 0
+95
View File
@@ -0,0 +1,95 @@
"""Task 3 — MonitoringQueryService: read-only filter/sort/pagination over
audit events, fully testable without file I/O (InMemoryAuditEventRepository)
and with a real CanonicalAuditLogger wired through CanonicalAuditEventRepository.
"""
from __future__ import annotations
from cowork_local.application.monitoring.dto.audit_event_dto import AuditEventDTO
from cowork_local.application.monitoring.monitoring_query_service import (
MonitoringQueryService,
)
from cowork_local.application.monitoring.repository.audit_event_repository import (
CanonicalAuditEventRepository,
InMemoryAuditEventRepository,
)
from cowork_local.infrastructure.telemetry.audit_logger import CanonicalAuditLogger
def _event(ts, kind="tool_call", name="x", ok=True, detail="") -> AuditEventDTO:
return AuditEventDTO(ts=ts, kind=kind, name=name, ok=ok, detail=detail)
def test_query_filters_by_kind() -> None:
repo = InMemoryAuditEventRepository([
_event("2026-01-01T00:00:00", kind="mcp_call", name="a"),
_event("2026-01-01T00:00:01", kind="security_block", name="b"),
])
service = MonitoringQueryService(repo)
page = service.query(kind="mcp_call")
assert [e.name for e in page.items] == ["a"]
def test_query_filters_by_ok_and_text() -> None:
repo = InMemoryAuditEventRepository([
_event("2026-01-01T00:00:00", name="run_command", ok=False, detail="blocked"),
_event("2026-01-01T00:00:01", name="run_command", ok=True, detail="fine"),
_event("2026-01-01T00:00:02", name="fetch_url", ok=False, detail="blocked"),
])
service = MonitoringQueryService(repo)
page = service.query(ok=False, text="run_command")
assert len(page.items) == 1
assert page.items[0].detail == "blocked"
def test_query_sorts_newest_first_by_default() -> None:
repo = InMemoryAuditEventRepository([
_event("2026-01-01T00:00:00", name="first"),
_event("2026-01-02T00:00:00", name="second"),
])
service = MonitoringQueryService(repo)
page = service.query()
assert [e.name for e in page.items] == ["second", "first"]
def test_query_paginates() -> None:
events = [_event(f"2026-01-{i:02d}T00:00:00", name=str(i)) for i in range(1, 11)]
repo = InMemoryAuditEventRepository(events)
service = MonitoringQueryService(repo)
page1 = service.query(sort_by="ts", descending=False, page=1, page_size=4)
page2 = service.query(sort_by="ts", descending=False, page=2, page_size=4)
assert page1.total == 10
assert [e.name for e in page1.items] == ["1", "2", "3", "4"]
assert [e.name for e in page2.items] == ["5", "6", "7", "8"]
assert page1.has_more is True
def test_large_page_size_returns_everything_matching_current_ui_behaviour() -> None:
events = [_event(f"2026-01-{i:02d}T00:00:00", name=str(i)) for i in range(1, 6)]
repo = InMemoryAuditEventRepository(events)
service = MonitoringQueryService(repo)
page = service.query(page_size=10_000)
assert len(page.items) == 5
assert page.has_more is False
def test_repository_is_read_only_no_pyside6_import() -> None:
import cowork_local.application.monitoring.monitoring_query_service as mod
import cowork_local.application.monitoring.repository.audit_event_repository as repo_mod
assert "PySide6" not in mod.__dict__
assert "PySide6" not in repo_mod.__dict__
assert not hasattr(mod.MonitoringQueryService, "record")
def test_canonical_repository_wires_to_real_logger(tmp_path) -> None:
logger = CanonicalAuditLogger(tmp_path)
logger.record("security_block", "run_command", False, detail="nope")
logger.record("mcp_call", "search", True)
repo = CanonicalAuditEventRepository(logger)
service = MonitoringQueryService(repo)
page = service.query(kind="security_block")
assert len(page.items) == 1
assert page.items[0].name == "run_command"
@@ -0,0 +1,73 @@
"""Task 1 (sub-step 6e) — SandboxDetailsCard / PermissionsCard.
Verifies the Permissions card is nested INSIDE the Sandbox card's fold
(matching the pre-refactor layout exactly) and that refresh()/retranslate()
compute the same values the original MonitoringTab methods did.
"""
from __future__ import annotations
import copy
import os
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
QApplication = pytest.importorskip("PySide6.QtWidgets").QApplication
from cowork_local.config import AppConfig, DEFAULT_CONFIG
from cowork_local.presentation.monitoring.tabs.sandbox_tab import SandboxDetailsCard
@pytest.fixture(scope="module")
def qapp():
app = QApplication.instance() or QApplication([])
yield app
class _FakeCtx:
def __init__(self, tmp_path, block_network=False):
self.config = AppConfig(data=copy.deepcopy(DEFAULT_CONFIG), path=tmp_path / "c.json")
self.config.data["agent_security"]["block_network"] = block_network
self.config.data["agent_security"]["resource_limit_cpu_percent"] = 50
import time
self.started_at = time.time() - 65 # ~1m5s uptime
def test_permissions_card_nested_inside_sandbox_fold(qapp, tmp_path) -> None:
card = SandboxDetailsCard(_FakeCtx(tmp_path), on_settings_changed=lambda: None)
assert card.permissions_card.parent() is card._detail
def test_refresh_computes_uptime_and_resource_limits(qapp, tmp_path) -> None:
card = SandboxDetailsCard(_FakeCtx(tmp_path), on_settings_changed=lambda: None)
card.retranslate()
card.refresh()
assert "CPU 50%" in card.limits_lbl.text()
assert "m" in card.uptime_val.text()
def test_refresh_reflects_network_blocked_on_both_cards(qapp, tmp_path) -> None:
card = SandboxDetailsCard(_FakeCtx(tmp_path, block_network=True), on_settings_changed=lambda: None)
card.retranslate()
card.refresh()
assert card.net_val.objectName() == "badgeWarn"
assert card.permissions_card.network_val.objectName() == "badgeWarn"
def test_refresh_reflects_network_allowed(qapp, tmp_path) -> None:
card = SandboxDetailsCard(_FakeCtx(tmp_path, block_network=False), on_settings_changed=lambda: None)
card.retranslate()
card.refresh()
assert card.net_val.objectName() == "badgeSuccess"
assert card.permissions_card.network_val.objectName() == "badgeSuccess"
def test_fold_starts_collapsed(qapp, tmp_path) -> None:
# isVisible() alone can't tell (it also depends on ancestors actually
# being shown on screen, which nothing here is) — isHidden() reflects
# the explicit setVisible(False) call regardless of the parent chain.
card = SandboxDetailsCard(_FakeCtx(tmp_path), on_settings_changed=lambda: None)
assert card._detail.isHidden() is True
card.more_btn.setChecked(True)
assert card._detail.isHidden() is False
+78
View File
@@ -0,0 +1,78 @@
"""Task 1 (sub-step 6a) — presentation/monitoring/shared helpers.
Only the pure-Python functions are covered here (no PySide6 widget
instantiation needed, no pytest-qt required). Values are asserted against
what ``ui/monitoring_tab.py``'s original, now-removed private functions
produced, so this doubles as a characterization test proving the extraction
didn't change output.
"""
from __future__ import annotations
from cowork_local.presentation.monitoring.shared import badges, formatters
def test_fmt_bytes() -> None:
assert formatters.fmt_bytes(500) == "500 B"
assert formatters.fmt_bytes(2048) == "2 KB"
def test_fmt_event_time_and_full_and_relative_are_unparsable_safe() -> None:
assert formatters.fmt_event_time("not-a-timestamp") == "not-a-timestamp"
assert formatters.fmt_event_time_full("not-a-timestamp") == "not-a-timestamp"
assert formatters.relative_time("not-a-timestamp") == ""
def test_fmt_event_time_formats_valid_iso_timestamp() -> None:
assert formatters.fmt_event_time("2026-05-25T15:03:00") == "25/05 15:03"
# The separator is computed via datetime.strftime with a literal
# non-ASCII character in the format string, same as the pre-refactor
# ui/monitoring_tab.py code — on a Windows box whose locale ANSI codepage
# has no direct mapping for U+00B7 (MIDDLE DOT), strftime's encode/decode
# round trip through that codepage can substitute a different but
# visually similar character (observed: U+30FB on a ja_JP/cp932 locale).
# That behavior is unchanged by this refactor either way, so the test
# computes the expected separator the exact same way the implementation
# does, rather than assuming byte 0xB7 survives on every locale.
import datetime as _dt
expected_full = _dt.datetime(2026, 5, 25, 15, 3, 7).strftime(
f"%d/%m/%Y {formatters._MIDDLE_DOT} %H:%M:%S")
assert formatters.fmt_event_time_full("2026-05-25T15:03:07") == expected_full
def test_event_id_shape() -> None:
assert formatters.event_id("2026-05-25T15:03:00", 7) == "evt_202605251503_007"
def test_agent_initials() -> None:
assert formatters.agent_initials("Cowork Agent") == "CA"
assert formatters.agent_initials("graphrag") == "G"
def test_agent_avatar_colour_identity_mapping() -> None:
assert formatters.agent_avatar_colour("Security Agent") == "#D13438"
assert formatters.agent_avatar_colour("Cowork Agent") == "#0078D4"
assert formatters.agent_avatar_colour("unknown-agent") == "#0078D4"
def test_action_label_falls_back_to_raw_name_when_unmapped() -> None:
assert badges.action_label("some_custom_tool") == "some_custom_tool"
def test_status_info_security_block_unmapped_defaults_to_blocked() -> None:
tone, key = badges.status_info("some_new_rule", kind="security_block", ok=False)
assert (tone, key) == ("badgePurple", "monitoring.status_blocked")
def test_status_info_mcp_call_falls_back_to_ok_flag() -> None:
assert badges.status_info("search", kind="mcp_call", ok=True) == ("badgeSuccess", "monitoring.status_ok")
assert badges.status_info("search", kind="mcp_call", ok=False) == ("badgeDanger", "monitoring.status_failed")
def test_severity_info_dangerous_command_is_critical() -> None:
assert badges.severity_info("run_command") == ("badgeDanger", "monitoring.severity_critical")
def test_agent_badge_name_identity_mapping() -> None:
assert badges.agent_badge_name("Security Agent") == "badgeDanger"
assert badges.agent_badge_name("graphrag") == "badgePurple"
assert badges.agent_badge_name("unknown") == "badge"
+97
View File
@@ -0,0 +1,97 @@
"""Task 1 (sub-step 6g) — the new presentation/monitoring/monitoring_tab.py
container. Builds a real MonitoringTab against a real AppConfig/AppContext
(same convention as tests/routing/test_service.py: real config/context, only
the true external collaborators — here, none — get a fake), and verifies the
public API app.py depends on is intact: constructor signature,
``status_message`` signal, ``select_subtab``, ``nav_subtabs``,
``hide_tab_bar``.
"""
from __future__ import annotations
import copy
import os
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
QApplication = pytest.importorskip("PySide6.QtWidgets").QApplication
from cowork_local.config import AppConfig, DEFAULT_CONFIG
from cowork_local.presentation.monitoring.monitoring_tab import MonitoringTab
from cowork_local.state import AppContext
@pytest.fixture(scope="module")
def qapp():
app = QApplication.instance() or QApplication([])
yield app
@pytest.fixture()
def ctx(tmp_path):
config = AppConfig(data=copy.deepcopy(DEFAULT_CONFIG), path=tmp_path / "config.json")
return AppContext(config)
def test_constructor_accepts_apps_py_call_signature(qapp, ctx) -> None:
# app.py:499-502 calls MonitoringTab(self.ctx, cowork=..., structure=...,
# task_scheduler=...) — all optional besides ctx.
tab = MonitoringTab(ctx)
assert tab.tabs.count() >= 1
def test_status_message_signal_exists(qapp, ctx) -> None:
tab = MonitoringTab(ctx)
received = []
tab.status_message.connect(received.append)
tab.status_message.emit("hello")
assert received == ["hello"]
def test_select_subtab_and_nav_subtabs(qapp, ctx) -> None:
tab = MonitoringTab(ctx)
subtabs = tab.nav_subtabs()
assert len(subtabs) == tab.tabs.count()
tab.select_subtab(1)
assert tab.tabs.currentIndex() == 1
def test_hide_tab_bar_does_not_raise(qapp, ctx) -> None:
tab = MonitoringTab(ctx)
tab.hide_tab_bar()
def test_full_refresh_populates_event_tabs_via_query_service(qapp, ctx, monkeypatch) -> None:
from cowork_local.core import audit_log
audit_dir = ctx.config.path.parent / "audit"
monkeypatch.setattr(audit_log, "AUDIT_DIR", audit_dir)
monkeypatch.setattr(audit_log, "_logger", audit_log.CanonicalAuditLogger(audit_dir))
audit_log.record("security_block", "run_command", False, detail="blocked")
audit_log.record("mcp_call", "search", True)
audit_log.record("tool_call", "read_file", True)
tab = MonitoringTab(ctx, cowork=None, structure=None, task_scheduler=None)
tab.refresh()
assert tab.security_tab.table.rowCount() == 1
assert tab.mcp_tab.table.rowCount() == 1
assert tab.action_tab.table.rowCount() == 3
def test_retranslate_does_not_raise(qapp, ctx) -> None:
tab = MonitoringTab(ctx)
tab._retranslate()
def test_settings_changed_callback_is_the_container_full_refresh(qapp, ctx) -> None:
# Editing Settings from either the Sandbox or the nested Permissions
# card's "Edit" button used to call the same
# MonitoringTab._open_settings_and_refresh -> self.refresh(); the
# container now wires both cards' on_settings_changed to its own
# bound refresh method, so this equality is the direct replacement
# for that identity.
tab = MonitoringTab(ctx)
assert tab.overview_tab.sandbox_card._on_settings_changed == tab.refresh
assert tab.overview_tab.sandbox_card.permissions_card._on_settings_changed == tab.refresh
+101
View File
@@ -0,0 +1,101 @@
"""Task 5 — Sandbox Capability Matrix: pure OS/risk-tier policy, no execution,
no PySide6, no dependency on core/sandbox_manager.py.
"""
from __future__ import annotations
from cowork_local.infrastructure.sandbox import sandbox_capabilities as sc
def test_detect_os_from_injected_platform_name() -> None:
assert sc.detect_os("win32") == sc.WINDOWS
assert sc.detect_os("linux") == sc.LINUX
assert sc.detect_os("darwin") == sc.MACOS
assert sc.detect_os("some-other-os") == sc.UNKNOWN
def test_windows_matches_todays_real_backends() -> None:
matrix = sc.SandboxCapabilityMatrix(operating_system=sc.WINDOWS)
names = {b.name for b in matrix.available_backends()}
assert names == {"direct", "integrity_job_wfp", "appcontainer", "windows_sandbox"}
def test_windows_routing_matches_core_sandbox_manager_today() -> None:
matrix = sc.SandboxCapabilityMatrix(operating_system=sc.WINDOWS)
assert matrix.select_backend(sc.SAFE) == "integrity_job_wfp"
assert matrix.select_backend(sc.MODERATE) == "integrity_job_wfp"
assert matrix.select_backend(sc.HIGH) == "appcontainer"
assert matrix.select_backend(sc.CRITICAL) == "windows_sandbox"
def test_linux_has_no_real_backend_yet() -> None:
matrix = sc.SandboxCapabilityMatrix(operating_system=sc.LINUX)
available = {b.name for b in matrix.available_backends()}
assert available == {"direct"} # namespaces_bubblewrap declared but not implemented
assert matrix.select_backend(sc.SAFE) == "direct" # SAFE/MODERATE only ever wanted direct
# HIGH's preferred backend (namespaces_bubblewrap) isn't implemented, and
# HIGH's routing table names "blocked" as the explicit next preference
# (not a silent fallback to unisolated "direct") — a HIGH-risk command
# must never quietly downgrade to no isolation just because the real
# sandbox backend is missing on this OS.
assert matrix.select_backend(sc.HIGH) == "blocked"
assert matrix.select_backend(sc.CRITICAL) == "blocked"
def test_macos_has_no_real_backend_yet() -> None:
matrix = sc.SandboxCapabilityMatrix(operating_system=sc.MACOS)
available = {b.name for b in matrix.available_backends()}
assert available == {"direct"}
def test_disallowing_direct_fallback_blocks_instead() -> None:
# LINUX's own HIGH routing already names "blocked" explicitly, so it
# doesn't exercise the allow_direct_fallback branch. Register a profile
# whose HIGH tier names only an unavailable backend (no explicit
# "direct"/"blocked" entry) to exercise the bottom-of-select_backend
# fallback path directly.
os_name = "test-os-fallback"
profile = sc.OsSandboxProfile(
operating_system=os_name,
backends=(sc.SandboxBackend("direct", "none", True),),
routing={sc.SAFE: ("direct",), sc.MODERATE: ("direct",),
sc.HIGH: ("not_yet_implemented",), sc.CRITICAL: ("not_yet_implemented",)},
)
sc.register_profile(profile)
try:
allowed = sc.SandboxCapabilityMatrix(operating_system=os_name, allow_direct_fallback=True)
disallowed = sc.SandboxCapabilityMatrix(operating_system=os_name, allow_direct_fallback=False)
assert allowed.select_backend(sc.HIGH) == "direct"
assert disallowed.select_backend(sc.HIGH) == "blocked"
# CRITICAL never falls back to direct even when allowed.
assert allowed.select_backend(sc.CRITICAL) == "blocked"
finally:
del sc._PROFILES[os_name]
def test_unknown_os_always_blocks() -> None:
matrix = sc.SandboxCapabilityMatrix(operating_system=sc.UNKNOWN)
assert matrix.available_backends() == ()
for tier in (sc.SAFE, sc.MODERATE, sc.HIGH, sc.CRITICAL):
assert matrix.select_backend(tier) == "blocked"
def test_registering_a_brand_new_os_requires_no_class_changes() -> None:
freebsd = "freebsd"
profile = sc.OsSandboxProfile(
operating_system=freebsd,
backends=(sc.SandboxBackend("direct", "none", True),),
routing={sc.SAFE: ("direct",), sc.MODERATE: ("direct",),
sc.HIGH: ("blocked",), sc.CRITICAL: ("blocked",)},
)
sc.register_profile(profile)
try:
matrix = sc.SandboxCapabilityMatrix(operating_system=freebsd)
assert matrix.select_backend(sc.SAFE) == "direct"
assert matrix.select_backend(sc.HIGH) == "blocked"
finally:
del sc._PROFILES[freebsd] # don't leak state into other tests
def test_no_pyside6_or_subprocess_dependency() -> None:
assert "PySide6" not in sc.__dict__
assert "subprocess" not in sc.__dict__
+9 -1541
View File
File diff suppressed because it is too large Load Diff