fix(operation-trace): speed up audit reads and align i18n locales

Performance of the operation traceability surface:

- Cache the retention / row-budget / dedupe policy per Workspace and
  invalidate it on governance change, so the hot write path no longer
  issues three metadata round trips per recorded operation.
- Resolve the governance payload with one metadata query instead of four.
- Turn the dedupe probe into an existence check (LIMIT 1) instead of a
  full COUNT over the dedupe window.
- Defer the row-budget COUNT to every N inserts; prune still enforces the
  exact budget, and the maintenance loop keeps it bounded.
- Panel: fetch only the records page when paging or filtering instead of
  re-requesting governance and the filter catalogue every time.
- Export: include the effective filters (and limit) in the artifact so a
  downloaded CSV is self-describing.

Fixes and cleanup:

- Drop the unused OperationLogPruneReport import that broke `tsc`.
- i18n: remove the 9 redundant blank lines unique to en-US and format the
  six non-compliant locale files with the project Prettier config; all
  eight locales keep identical key sets, key order and placeholders.
This commit is contained in:
TyperBody
2026-09-26 02:11:42 +08:00
parent 4b9a1590df
commit 78baabde96
25 changed files with 3893 additions and 13 deletions
+138 -4
View File
@@ -4,6 +4,7 @@ import abc
import typing
import enum
import quart
import time
import traceback
import inspect
import uuid
@@ -94,12 +95,23 @@ class RouterGroup(abc.ABC):
rule: str,
auth_type: AuthType = AuthType.USER_TOKEN,
permission: Permission | str | None = None,
operation_log_meta: dict[str, typing.Any] | None = None,
**options: typing.Any,
) -> typing.Callable[[RouteCallable], RouteCallable]: # decorator
"""Register a route"""
"""Register a route.
``operation_log_meta`` optionally enriches the traceability record for
this route with trusted server-side facts. Supported keys:
* ``resource_param``: handler kwarg holding the target resource id,
* ``changes``: a pre-computed before/after diff to persist,
* ``detail``: extra structured context (sensitive keys are redacted).
"""
if auth_type == AuthType.ACCOUNT_TOKEN and permission is not None:
raise ValueError('Account-token routes cannot declare Workspace permissions')
if operation_log_meta is not None and not isinstance(operation_log_meta, dict):
raise ValueError('operation_log_meta must be a dict')
def decorator(f: RouteCallable) -> RouteCallable:
nonlocal rule
@@ -223,6 +235,7 @@ class RouterGroup(abc.ABC):
except Exception as e:
return self._auth_error_response(e)
observed_at = time.monotonic()
try:
if request_context is not None:
diagnostics.workspace(request_context)
@@ -245,11 +258,40 @@ class RouterGroup(abc.ABC):
# runtimes, uploads, or streamed clients.
# Services that need atomic writes open a UoW.
async with persistence_mgr.tenant_scope(request_context.workspace_uuid):
return await f(*args, **kwargs)
return await f(*args, **kwargs)
return await f(*args, **kwargs)
response = await f(*args, **kwargs)
else:
response = await f(*args, **kwargs)
else:
response = await f(*args, **kwargs)
if request_context is not None:
# Trace after the handler released its tenant scope so
# auditing never extends a business transaction. The
# call is a no-op (and costs no database round trip)
# when tracing is off for this Workspace.
await self._record_operation(
request_context,
rule,
options,
kwargs,
observed_at,
response,
meta=operation_log_meta,
)
return response
except Exception as e: # 自动 500
if request_context is not None:
await self._record_operation(
request_context,
rule,
options,
kwargs,
observed_at,
None,
failure=e,
meta=operation_log_meta,
)
if isinstance(e, CodexProviderError):
return self.http_status(e.status_code, e.error_code, str(e))
if isinstance(e, AuthorizationError):
@@ -301,6 +343,98 @@ class RouterGroup(abc.ABC):
return decorator
async def _record_operation(
self,
ctx: RequestContext,
route: str,
options: dict[str, typing.Any],
kwargs: dict[str, typing.Any],
observed_at: float,
response: typing.Any,
*,
failure: BaseException | None = None,
meta: dict[str, typing.Any] | None = None,
) -> None:
"""Best-effort traceability for one finished request.
This runs after the handler released its tenant scope. It never raises:
an audit failure must not turn a successful business response into a
500. The service performs its own cheap level check before touching the
database, so a disabled Workspace pays nothing at steady state.
"""
service = getattr(self.ap, 'workspace_settings_service', None)
if service is None:
return
try:
method = str(quart.request.method or options.get('methods', ['GET'])[0])
status_code = self._extract_status_code(response, failure)
if failure is not None:
outcome = 'denied' if isinstance(failure, AuthorizationError) else 'error'
elif status_code >= 500:
outcome = 'error'
elif status_code >= 400:
outcome = 'denied'
else:
outcome = 'ok'
resolved_meta = meta or {}
resource_id = None
resource_param = resolved_meta.get('resource_param')
if isinstance(resource_param, str):
candidate = kwargs.get(resource_param)
if candidate is not None:
resource_id = str(candidate)
# Handlers may publish request-local traceability facts when a
# static route declaration cannot describe the runtime diff, e.g.
# the previous and the new role of a member. ``quart.g`` is
# request-local, so concurrent requests never share these values.
changes = resolved_meta.get('changes')
detail = resolved_meta.get('detail')
request_changes = getattr(quart.g, 'operation_log_changes', None)
request_detail = getattr(quart.g, 'operation_log_detail', None)
request_resource_id = getattr(quart.g, 'operation_log_resource_id', None)
request_summary = getattr(quart.g, 'operation_log_summary', None)
if request_changes:
changes = request_changes
if request_detail:
detail = request_detail
if request_resource_id:
resource_id = str(request_resource_id)
await service.record_request(
ctx,
method=method,
route=route,
status_code=status_code,
outcome=outcome,
duration_ms=int(max(time.monotonic() - observed_at, 0.0) * 1000),
resource_id=resource_id,
changes=changes or None,
detail=detail or None,
)
except Exception as exc: # pragma: no cover - auditing is best effort
logger = getattr(self.ap, 'logger', self.quart_app.logger)
logger.debug(f'Operation trace skipped for {route}: {exc}')
@staticmethod
def _extract_status_code(response: typing.Any, failure: BaseException | None) -> int:
"""Resolve the HTTP status of a handler result without consuming it."""
if failure is not None:
status = getattr(failure, 'status_code', None)
if isinstance(status, int):
return status
return 500
if isinstance(response, tuple) and len(response) >= 2 and isinstance(response[1], int):
return response[1]
status = getattr(response, 'status_code', None)
if isinstance(status, int):
return status
return 200
async def _authenticate_account(self, token: str) -> tuple[typing.Any, str]:
account: typing.Any = None
resolver = getattr(self.ap.user_service, 'get_authenticated_account', None)
@@ -0,0 +1,244 @@
"""Workspace governance settings routes.
Exposes the "成员操作日志溯源" (member operation traceability) surface:
* ``GET /api/v1/settings/governance`` read level + template table
* ``PUT /api/v1/settings/governance`` change level / retention
* ``GET /api/v1/settings/operation-logs`` page through records
* ``GET /api/v1/settings/operation-logs/filters`` available filter values
* ``POST /api/v1/settings/operation-logs/prune`` enforce retention now
* ``DEL /api/v1/settings/operation-logs`` clear records
Every route requires ``audit.view``, which is granted to the Workspace owner
and admin only. Write routes additionally require an owning/admin role so a
future permission grant cannot silently widen who may reconfigure tracing.
"""
from __future__ import annotations
import datetime
import json
import typing
import quart
from ...authz import Permission
from ...context import RequestContext
from ...service import settings as settings_service
from .. import group
def _parse_timestamp(value: str | None) -> datetime.datetime | None:
"""Parse an ISO-8601 query parameter, tolerating a trailing ``Z``."""
if not value:
return None
candidate = value.strip()
if not candidate:
return None
if candidate.endswith('Z'):
candidate = candidate[:-1] + '+00:00'
try:
parsed = datetime.datetime.fromisoformat(candidate)
except ValueError:
return None
# Stored timestamps are naive UTC; normalise aware inputs before comparing.
if parsed.tzinfo is not None:
parsed = parsed.astimezone(datetime.timezone.utc).replace(tzinfo=None)
return parsed
def _parse_int(value: str | None, default: int) -> int:
try:
return int(value) if value is not None else default
except (TypeError, ValueError):
return default
def _csv_cell(value: typing.Any) -> str:
"""Render one CSV cell with the standard quoting rules."""
if value is None:
return ''
text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, default=str)
return '"' + text.replace('"', '""') + '"'
@group.group_class('settings', '/api/v1/settings')
class SettingsRouterGroup(group.RouterGroup):
async def initialize(self) -> None:
@self.route(
'/governance',
methods=['GET', 'PUT'],
auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
permission=Permission.AUDIT_VIEW,
)
async def _(request_context: RequestContext) -> typing.Any:
service = self.ap.workspace_settings_service
if quart.request.method == 'GET':
return self.success(data=await service.describe_governance(request_context.workspace_uuid))
if not settings_service.role_may_configure(request_context.workspace.role):
return self.http_status(
403,
'permission_denied',
'Only Workspace owners and admins can change tracing settings',
)
data = await quart.request.get_json(silent=True) or {}
# The previous level is captured before the write so the trace can
# answer "what was changed into what" for the governance change
# itself, not only for the resources it protects.
before = await service.describe_governance(request_context.workspace_uuid)
updated = await service.set_operation_level(
request_context.workspace_uuid,
data.get('level', before['configured_level']),
retention_days=data.get('retention_days'),
max_rows=data.get('max_rows'),
dedupe_window_seconds=data.get('dedupe_window_seconds'),
)
account_name = None
if request_context.account_uuid and self.ap.user_service is not None:
account = await self.ap.user_service.get_user_by_uuid(request_context.account_uuid)
account_name = account.user if account is not None else None
changes = settings_service.changed_fields(
{
'configured_level': before['configured_level'],
'retention_days': before['retention_days'],
'max_rows': before['max_rows'],
'dedupe_window_seconds': before['dedupe_window_seconds'],
},
{
'configured_level': updated['configured_level'],
'retention_days': updated['retention_days'],
'max_rows': updated['max_rows'],
'dedupe_window_seconds': updated['dedupe_window_seconds'],
},
)
if changes:
await service.record(
request_context.workspace_uuid,
rule=settings_service.ACTION_RULES_BY_ACTION['settings_update'],
level=max(
await service.effective_level(
request_context.workspace_uuid,
request_context.workspace.role,
),
settings_service.OPERATION_LEVEL_MUTATION,
),
actor_account_uuid=request_context.account_uuid,
actor_name=account_name,
actor_role=request_context.workspace.role,
principal_type=request_context.principal.principal_type.value,
request_id=request_context.request_id,
http_method=quart.request.method,
route='/api/v1/settings/governance',
outcome='ok',
summary=settings_service.build_summary(
settings_service.ACTION_RULES_BY_ACTION['settings_update'], changes
),
changes=changes,
)
return self.success(data=updated)
@self.route(
'/operation-logs',
methods=['GET'],
auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
permission=Permission.AUDIT_VIEW,
)
async def _(request_context: RequestContext) -> typing.Any:
# Read-only by design: operation records are immutable and cannot
# be deleted on demand. Only the retention policy may prune them.
service = self.ap.workspace_settings_service
args = quart.request.args
level_raw = args.get('level')
result = await service.query_logs(
request_context.workspace_uuid,
limit=_parse_int(args.get('limit'), settings_service.DEFAULT_PAGE_SIZE),
offset=_parse_int(args.get('offset'), 0),
action=args.get('action') or None,
resource_type=args.get('resource_type') or None,
actor_account_uuid=args.get('actor') or None,
level=_parse_int(level_raw, 0) if level_raw not in (None, '') else None,
since=_parse_timestamp(args.get('since')),
until=_parse_timestamp(args.get('until')),
)
return self.success(data=result)
@self.route(
'/operation-logs/filters',
methods=['GET'],
auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
permission=Permission.AUDIT_VIEW,
)
async def _(request_context: RequestContext) -> typing.Any:
service = self.ap.workspace_settings_service
options = await service.list_filter_options(request_context.workspace_uuid)
options['governance'] = await service.describe_governance(request_context.workspace_uuid)
return self.success(data=options)
@self.route(
'/operation-logs/export',
methods=['GET'],
auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
permission=Permission.AUDIT_VIEW,
)
async def _(request_context: RequestContext) -> typing.Any:
"""Download the filtered records as a CSV attachment.
The export is read-only and shares the exact filters of the panel,
so what an operator sees is what the artifact contains.
"""
args = quart.request.args
level_raw = args.get('level')
service = self.ap.workspace_settings_service
document = await service.export_logs(
request_context.workspace_uuid,
action=args.get('action') or None,
resource_type=args.get('resource_type') or None,
actor_account_uuid=args.get('actor') or None,
level=_parse_int(level_raw, 0) if level_raw not in (None, '') else None,
since=_parse_timestamp(args.get('since')),
until=_parse_timestamp(args.get('until')),
)
columns = (
'id',
'created_at',
'action',
'resource_type',
'resource_id',
'actor_account_uuid',
'actor_name',
'actor_role',
'principal_type',
'http_method',
'route',
'outcome',
'status_code',
'level_name',
'integrity_ok',
'chain_ok',
'tampered',
'record_hash',
'prev_hash',
'duration_ms',
'summary',
'changes',
'detail',
)
lines = [','.join(columns)]
for record in document['records']:
lines.append(','.join(_csv_cell(record.get(column)) for column in columns))
# Prepend a UTF-8 BOM so spreadsheet tools detect the encoding.
body = '\ufeff' + '\r\n'.join(lines) + '\r\n'
filename = f"operation-logs-{document['exported_at'].replace(':', '')}.csv"
response = quart.Response(body, mimetype='text/csv', content_type='text/csv; charset=utf-8')
response.headers['Content-Disposition'] = f'attachment; filename="{filename}"'
return response
@@ -12,6 +12,7 @@ from .....entity.persistence.workspace import WorkspaceSource
from .....workspace.collaboration import WorkspaceMemberView
from .....workspace.errors import WorkspaceNotFoundError
from .....workspace.invitation_delivery import InvitationDeliveryService
from ...service import settings as settings_service
from .. import group
@@ -218,6 +219,18 @@ class WorkspacesRouterGroup(group.RouterGroup):
)
created = await self._run_in_workspace_uow(workspace_uuid, create_invitation)
invite_rule = settings_service.ACTION_RULES_BY_ACTION['member_invite']
invite_changes = [
{
'field': 'invitation',
'before': None,
'after': f"{created.invitation.normalized_email}:{created.invitation.role}",
}
]
quart.g.operation_log_changes = invite_changes
quart.g.operation_log_resource_id = created.invitation.uuid
quart.g.operation_log_detail = {'role': created.invitation.role}
quart.g.operation_log_summary = settings_service.build_summary(invite_rule, invite_changes)
delivery_service = self._invitation_delivery_service()
link = delivery_service.build_invitation_link(created.token)
workspace = await self.ap.workspace_service.get_workspace(workspace_uuid)
@@ -266,6 +279,11 @@ class WorkspacesRouterGroup(group.RouterGroup):
request_context: RequestContext,
) -> typing.Any:
self._require_current_workspace(workspace_uuid, request_context)
# Capture the previous role before mutating so the operation trace
# can answer "what was changed into what" for this member.
before_role = await self._member_role(workspace_uuid, account_uuid)
if quart.request.method == 'DELETE':
if Permission.MEMBER_REMOVE.value not in request_context.workspace.permissions:
return self.http_status(403, 'permission_denied', 'Member removal permission is required')
@@ -276,6 +294,17 @@ class WorkspacesRouterGroup(group.RouterGroup):
)
member = await self._run_in_workspace_uow(workspace_uuid, remove_member)
await self._publish_member_trace(
rule_action='member_remove',
account_uuid=account_uuid,
changes=[
{
'field': 'member',
'before': f'{account_uuid}:{before_role or "unknown"}',
'after': 'removed',
}
],
)
return self.success(data={'account_uuid': member.account_uuid})
data = await quart.request.get_json(silent=True) or {}
@@ -290,6 +319,18 @@ class WorkspacesRouterGroup(group.RouterGroup):
member = await self._run_in_workspace_uow(workspace_uuid, update_member_role)
account = await self.ap.user_service.get_user_by_uuid(member.account_uuid)
role_changes = [
{'field': 'role', 'before': before_role, 'after': member.role},
]
await self._publish_member_trace(
rule_action='member_role_update',
account_uuid=account_uuid,
changes=role_changes,
detail={
'member_email': account.normalized_email if account is not None else None,
},
)
return self.success(
data={
'member': _membership_payload(
@@ -300,6 +341,41 @@ class WorkspacesRouterGroup(group.RouterGroup):
}
)
async def _member_role(self, workspace_uuid: str, account_uuid: str) -> str | None:
"""Return a member's current role without failing the request."""
try:
members = await self.ap.workspace_collaboration_service.list_members(
workspace_uuid, quart.g.workspace_membership
)
except Exception:
return None
for view in members:
if view.membership.account_uuid == account_uuid:
return view.membership.role
return None
async def _publish_member_trace(
self,
*,
rule_action: str,
account_uuid: str,
changes: list[dict[str, typing.Any]],
detail: dict[str, typing.Any] | None = None,
) -> None:
"""Publish request-local trace facts consumed by the route wrapper.
``quart.g`` is request-local, so concurrent member operations never
share these values.
"""
rule = settings_service.ACTION_RULES_BY_ACTION.get(rule_action)
quart.g.operation_log_changes = changes
quart.g.operation_log_resource_id = account_uuid
quart.g.operation_log_detail = detail or {}
if rule is not None:
quart.g.operation_log_summary = settings_service.build_summary(rule, changes)
@staticmethod
def _require_current_workspace(workspace_uuid: str, request_context: RequestContext) -> None:
if workspace_uuid != request_context.workspace_uuid:
File diff suppressed because it is too large Load Diff
+41
View File
@@ -36,6 +36,7 @@ from ..api.http.service import mcp as mcp_service
from ..api.http.service import apikey as apikey_service
from ..api.http.service import webhook as webhook_service
from ..api.http.service import monitoring as monitoring_service
from ..api.http.service import settings as workspace_settings_service
from ..api.http.service import skill as skill_service
from ..api.http.service import maintenance as maintenance_service
from ..discover import engine as discover_engine
@@ -198,6 +199,8 @@ class Application:
monitoring_service: monitoring_service.MonitoringService = None
workspace_settings_service: workspace_settings_service.WorkspaceSettingsService = None
skill_service: skill_service.SkillService = None
skill_mgr: skill_mgr.SkillManager = None
@@ -434,6 +437,24 @@ class Application:
* 3600
)
# Operation-log retention piggybacks on the shared maintenance loop
# instead of starting another long-lived task, so enabling
# traceability adds no scheduling cost of its own.
operation_log_cleanup_cfg = (
self.instance_config.data.get('operation_log', {}).get('auto_cleanup', {})
)
operation_log_enabled = (
operation_log_cleanup_cfg.get('enabled', True) and self.workspace_settings_service is not None
)
operation_log_interval_seconds = (
self._get_positive_float_config(
operation_log_cleanup_cfg.get('check_interval_hours', 1),
default=1,
name='operation_log.auto_cleanup.check_interval_hours',
)
* 3600
)
maintenance_intervals: dict[str, float] = {}
if monitoring_enabled:
maintenance_intervals['monitoring'] = monitoring_interval_seconds
@@ -441,6 +462,8 @@ class Application:
maintenance_intervals['storage'] = storage_interval_seconds
if self.workspace_collaboration_service is not None:
maintenance_intervals['invitations'] = 3600.0
if operation_log_enabled:
maintenance_intervals['operation_logs'] = operation_log_interval_seconds
if maintenance_intervals:
@@ -506,6 +529,24 @@ class Application:
self.logger.warning(
f'Storage maintenance failed for Workspace {context.workspace_uuid}: {exc}'
)
if 'operation_logs' in due:
try:
report = await self.workspace_settings_service.prune(
context.workspace_uuid,
)
removed = report['expired'] + report['trimmed']
if removed > 0:
self.logger.info(
f'Operation log auto-cleanup for Workspace '
f'{context.workspace_uuid}: removed {removed} records'
)
except asyncio.CancelledError:
raise
except Exception as exc:
self.logger.warning(
f'Operation log auto-cleanup failed for '
f'Workspace {context.workspace_uuid}: {exc}'
)
if 'invitations' in due:
try:
await self.workspace_collaboration_service.cleanup_expired_invitations(
+4
View File
@@ -29,6 +29,7 @@ from ...api.http.service import mcp as mcp_service
from ...api.http.service import apikey as apikey_service
from ...api.http.service import webhook as webhook_service
from ...api.http.service import monitoring as monitoring_service
from ...api.http.service import settings as settings_service
from ...api.http.service import skill as skill_service
from ...skill import manager as skill_mgr
from ...api.http.service import maintenance as maintenance_service
@@ -127,6 +128,9 @@ class BuildAppStage(stage.BootingStage):
mcp_service_inst = mcp_service.MCPService(ap)
ap.mcp_service = mcp_service_inst
workspace_settings_service_inst = settings_service.WorkspaceSettingsService(ap)
ap.workspace_settings_service = workspace_settings_service_inst
apikey_service_inst = apikey_service.ApiKeyService(ap)
ap.apikey_service = apikey_service_inst
@@ -0,0 +1,179 @@
"""Workspace operation log entity for member action traceability.
This table is the append-only fact source for "who did what to which
resource" inside one Workspace. It is deliberately separate from
:mod:`event_log` (Runner/runtime facts) and from monitoring tables
(LLM/tool telemetry) so that administrative traceability can be enabled,
queried and pruned without touching the hot runtime path.
Design constraints
------------------
* One row per recorded operation; no updates after insert (append-only).
* ``changes`` holds the before/after diff as a compact JSON string so the
log answers the question "what was changed into what".
* Rows are tenant-owned: every row carries ``workspace_uuid`` and is
covered by the shared Row Level Security contract.
"""
from __future__ import annotations
import datetime
import sqlalchemy
from .base import Base
# Operation levels, ordered by increasing capture scope. They live in the
# entity module so both the persistence layer and the HTTP layer can import
# one single source of truth without a circular dependency.
OPERATION_LEVEL_NONE = 0
"""Record nothing (tracing disabled for the Workspace)."""
OPERATION_LEVEL_MUTATION = 1
"""Record create / update / delete / role changes only."""
OPERATION_LEVEL_READ = 2
"""Additionally record every read / view / list operation."""
OPERATION_LEVEL_NAMES = {
OPERATION_LEVEL_NONE: 'off',
OPERATION_LEVEL_MUTATION: 'mutation',
OPERATION_LEVEL_READ: 'read',
}
class WorkspaceOperationLog(Base):
"""One traceable administrative operation inside a Workspace."""
__tablename__ = 'workspace_operation_logs'
id = sqlalchemy.Column(sqlalchemy.BigInteger().with_variant(sqlalchemy.Integer, 'sqlite'), primary_key=True, autoincrement=True)
workspace_uuid = sqlalchemy.Column(
sqlalchemy.String(36),
sqlalchemy.ForeignKey('workspaces.uuid', ondelete='CASCADE'),
nullable=False,
index=True,
)
"""Owning Workspace; also the tenant isolation key."""
# --- Actor: who performed the operation -----------------------------
actor_account_uuid = sqlalchemy.Column(sqlalchemy.String(36), nullable=True, index=True)
"""Account UUID of the actor, when the principal is an Account."""
actor_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
"""Human readable actor identifier captured at write time."""
actor_role = sqlalchemy.Column(sqlalchemy.String(32), nullable=True)
"""Workspace role the actor held while performing the operation."""
principal_type = sqlalchemy.Column(sqlalchemy.String(32), nullable=True)
"""Principal kind (account, api_key, support_admin, system)."""
api_key_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
"""API key UUID when the operation was driven by an API key."""
auth_type = sqlalchemy.Column(sqlalchemy.String(32), nullable=True)
"""Authentication mode used for the request."""
# --- Request correlation --------------------------------------------
request_id = sqlalchemy.Column(sqlalchemy.String(128), nullable=True, index=True)
"""Stable request id shared with logs and error responses."""
http_method = sqlalchemy.Column(sqlalchemy.String(12), nullable=True)
"""HTTP verb of the traced request."""
route = sqlalchemy.Column(sqlalchemy.String(512), nullable=True, index=True)
"""Registered Core route identity (never a raw user-supplied URL)."""
# --- What happened ---------------------------------------------------
action = sqlalchemy.Column(sqlalchemy.String(64), nullable=True, index=True)
"""Normalized action verb (create, update, delete, view, ...)."""
resource_type = sqlalchemy.Column(sqlalchemy.String(64), nullable=True, index=True)
"""Resource family the operation targeted (member, resource, ...)."""
resource_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
"""Identifier of the targeted resource, when known."""
level = sqlalchemy.Column(sqlalchemy.Integer, nullable=False, server_default='1', index=True)
"""Capture level that produced this row (1 = mutation, 2 = read)."""
outcome = sqlalchemy.Column(sqlalchemy.String(16), nullable=False, server_default='ok')
"""Result bucket: ok, denied or error."""
status_code = sqlalchemy.Column(sqlalchemy.Integer, nullable=True)
"""HTTP status code when the operation surfaced one."""
# --- Payloads --------------------------------------------------------
summary = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
"""Short human readable description of the operation."""
changes = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
"""Before/after diff as a JSON string: what was changed into what."""
detail = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
"""Additional structured context as a JSON string."""
# --- Tamper evidence (append-only hash chain) ------------------------
record_hash = sqlalchemy.Column(sqlalchemy.String(64), nullable=True)
"""Keyed hash over this row's canonical content.
Recomputed on read: a mismatch marks the card as tampered.
"""
prev_hash = sqlalchemy.Column(sqlalchemy.String(64), nullable=True)
"""Previous record's ``record_hash`` in the same Workspace.
Linking rows this way makes silent edits or deletions detectable.
"""
dedupe_key = sqlalchemy.Column(sqlalchemy.String(64), nullable=True, index=True)
"""Hash used to collapse repeated read observations into one row."""
# --- Client fingerprint ---------------------------------------------
client_ip = sqlalchemy.Column(sqlalchemy.String(64), nullable=True)
"""Best-effort client address."""
user_agent = sqlalchemy.Column(sqlalchemy.String(512), nullable=True)
"""Client user agent, truncated before insert."""
duration_ms = sqlalchemy.Column(sqlalchemy.Integer, nullable=False, server_default='0')
"""Handler wall-clock duration in milliseconds."""
created_at = sqlalchemy.Column(
sqlalchemy.DateTime,
nullable=False,
server_default=sqlalchemy.func.now(),
index=True,
)
"""When the operation was recorded."""
__table_args__ = (
sqlalchemy.Index(
'ix_workspace_operation_logs_workspace_created',
'workspace_uuid',
'created_at',
),
sqlalchemy.Index(
'ix_workspace_operation_logs_workspace_resource',
'workspace_uuid',
'resource_type',
'created_at',
),
sqlalchemy.Index(
'ix_workspace_operation_logs_workspace_dedupe',
'workspace_uuid',
'dedupe_key',
'created_at',
),
sqlalchemy.CheckConstraint(
'level IN (0, 1, 2)',
name='ck_workspace_operation_logs_level',
),
sqlalchemy.CheckConstraint(
"outcome IN ('ok', 'denied', 'error')",
name='ck_workspace_operation_logs_outcome',
),
)
@@ -28,6 +28,7 @@ from langbot.pkg.entity.persistence import (
metadata, # noqa: F401
model, # noqa: F401
monitoring, # noqa: F401
operation_log, # noqa: F401
pipeline, # noqa: F401
plugin, # noqa: F401
rag, # noqa: F401
@@ -0,0 +1,133 @@
"""add workspace operation logs for member/resource traceability
Revision ID: 0032_workspace_operation_logs
Revises: 0031_merge_totp_assistant
Create Date: 2026-09-25
The table is append-only and tenant-owned. Fresh installs already receive it
from SQLAlchemy ``create_all``; this revision makes the change safe for
existing databases and mirrors the shared Row Level Security contract so the
Cloud release validation can enforce it.
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
revision = '0032_workspace_operation_logs'
down_revision = '0031_merge_totp_assistant'
branch_labels = None
depends_on = None
_TABLE_NAME = 'workspace_operation_logs'
_POLICY_NAME = 'langbot_workspace_isolation'
_TENANT_SETTING = 'langbot.workspace_uuid'
def _setting(name: str) -> str:
return f"NULLIF(current_setting('{name}', true), '')"
def _quote(conn: sa.Connection, identifier: str) -> str:
return conn.dialect.identifier_preparer.quote(identifier)
def upgrade() -> None:
conn = op.get_bind()
existing_tables = set(sa.inspect(conn).get_table_names())
if _TABLE_NAME not in existing_tables:
op.create_table(
_TABLE_NAME,
sa.Column('id', sa.BigInteger().with_variant(sa.Integer, 'sqlite'), nullable=False),
sa.Column(
'workspace_uuid',
sa.String(36),
sa.ForeignKey('workspaces.uuid', ondelete='CASCADE'),
nullable=False,
),
sa.Column('actor_account_uuid', sa.String(36), nullable=True),
sa.Column('actor_name', sa.String(255), nullable=True),
sa.Column('actor_role', sa.String(32), nullable=True),
sa.Column('principal_type', sa.String(32), nullable=True),
sa.Column('api_key_uuid', sa.String(255), nullable=True),
sa.Column('auth_type', sa.String(32), nullable=True),
sa.Column('request_id', sa.String(128), nullable=True),
sa.Column('http_method', sa.String(12), nullable=True),
sa.Column('route', sa.String(512), nullable=True),
sa.Column('action', sa.String(64), nullable=True),
sa.Column('resource_type', sa.String(64), nullable=True),
sa.Column('resource_id', sa.String(255), nullable=True),
sa.Column('level', sa.Integer(), nullable=False, server_default='1'),
sa.Column('outcome', sa.String(16), nullable=False, server_default='ok'),
sa.Column('status_code', sa.Integer(), nullable=True),
sa.Column('summary', sa.Text(), nullable=True),
sa.Column('changes', sa.Text(), nullable=True),
sa.Column('detail', sa.Text(), nullable=True),
sa.Column('client_ip', sa.String(64), nullable=True),
sa.Column('user_agent', sa.String(512), nullable=True),
sa.Column('duration_ms', sa.Integer(), nullable=False, server_default='0'),
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
sa.PrimaryKeyConstraint('id'),
sa.CheckConstraint('level IN (0, 1, 2)', name='ck_workspace_operation_logs_level'),
sa.CheckConstraint(
"outcome IN ('ok', 'denied', 'error')",
name='ck_workspace_operation_logs_outcome',
),
)
op.create_index(
'ix_workspace_operation_logs_workspace_created',
_TABLE_NAME,
['workspace_uuid', 'created_at'],
unique=False,
)
op.create_index(
'ix_workspace_operation_logs_workspace_resource',
_TABLE_NAME,
['workspace_uuid', 'resource_type', 'created_at'],
unique=False,
)
op.create_index(
'ix_workspace_operation_logs_workspace_uuid',
_TABLE_NAME,
['workspace_uuid'],
unique=False,
)
op.create_index('ix_workspace_operation_logs_created_at', _TABLE_NAME, ['created_at'], unique=False)
op.create_index('ix_workspace_operation_logs_action', _TABLE_NAME, ['action'], unique=False)
op.create_index('ix_workspace_operation_logs_resource_type', _TABLE_NAME, ['resource_type'], unique=False)
op.create_index('ix_workspace_operation_logs_level', _TABLE_NAME, ['level'], unique=False)
if conn.dialect.name != 'postgresql':
return
table = _quote(conn, _TABLE_NAME)
policy = _quote(conn, _POLICY_NAME)
expression = f'workspace_uuid::text = {_setting(_TENANT_SETTING)}'
op.execute(sa.text(f'ALTER TABLE {table} ENABLE ROW LEVEL SECURITY'))
op.execute(sa.text(f'ALTER TABLE {table} FORCE ROW LEVEL SECURITY'))
op.execute(sa.text(f'DROP POLICY IF EXISTS {policy} ON {table}'))
op.execute(
sa.text(
f'CREATE POLICY {policy} ON {table} AS PERMISSIVE FOR ALL TO PUBLIC '
f'USING ({expression}) WITH CHECK ({expression})'
)
)
def downgrade() -> None:
conn = op.get_bind()
if conn.dialect.name == 'postgresql':
table = _quote(conn, _TABLE_NAME)
policy = _quote(conn, _POLICY_NAME)
op.execute(sa.text(f'DROP POLICY IF EXISTS {policy} ON {table}'))
op.drop_index('ix_workspace_operation_logs_level', table_name=_TABLE_NAME)
op.drop_index('ix_workspace_operation_logs_resource_type', table_name=_TABLE_NAME)
op.drop_index('ix_workspace_operation_logs_action', table_name=_TABLE_NAME)
op.drop_index('ix_workspace_operation_logs_created_at', table_name=_TABLE_NAME)
op.drop_index('ix_workspace_operation_logs_workspace_uuid', table_name=_TABLE_NAME)
op.drop_index('ix_workspace_operation_logs_workspace_resource', table_name=_TABLE_NAME)
op.drop_index('ix_workspace_operation_logs_workspace_created', table_name=_TABLE_NAME)
op.drop_table(_TABLE_NAME)
@@ -0,0 +1,70 @@
"""add tamper-evidence columns to workspace operation logs
Revision ID: 0033_operation_log_integrity
Revises: 0032_workspace_operation_logs
Create Date: 2026-09-25
Adds the append-only hash chain (``record_hash`` / ``prev_hash``) and the
``dedupe_key`` used to collapse repeated read observations. Fresh databases
already receive these columns from SQLAlchemy ``create_all``; this revision
upgrades existing databases and is guarded so neither path fails.
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
revision = '0033_operation_log_integrity'
down_revision = '0032_workspace_operation_logs'
branch_labels = None
depends_on = None
_TABLE_NAME = 'workspace_operation_logs'
_NEW_COLUMNS = (
sa.Column('record_hash', sa.String(64), nullable=True),
sa.Column('prev_hash', sa.String(64), nullable=True),
sa.Column('dedupe_key', sa.String(64), nullable=True),
)
_DEDUPE_INDEX = 'ix_workspace_operation_logs_workspace_dedupe'
def upgrade() -> None:
conn = op.get_bind()
if _TABLE_NAME not in set(sa.inspect(conn).get_table_names()):
return
existing = {column['name'] for column in sa.inspect(conn).get_columns(_TABLE_NAME)}
missing = [column for column in _NEW_COLUMNS if column.name not in existing]
if missing:
with op.batch_alter_table(_TABLE_NAME, recreate='auto') as batch:
for column in missing:
batch.add_column(column)
index_names = {index['name'] for index in sa.inspect(conn).get_indexes(_TABLE_NAME)}
if _DEDUPE_INDEX not in index_names:
op.create_index(
_DEDUPE_INDEX,
_TABLE_NAME,
['workspace_uuid', 'dedupe_key', 'created_at'],
unique=False,
)
def downgrade() -> None:
conn = op.get_bind()
if _TABLE_NAME not in set(sa.inspect(conn).get_table_names()):
return
index_names = {index['name'] for index in sa.inspect(conn).get_indexes(_TABLE_NAME)}
if _DEDUPE_INDEX in index_names:
op.drop_index(_DEDUPE_INDEX, table_name=_TABLE_NAME)
existing = {column['name'] for column in sa.inspect(conn).get_columns(_TABLE_NAME)}
removable = [column.name for column in _NEW_COLUMNS if column.name in existing]
if removable:
with op.batch_alter_table(_TABLE_NAME, recreate='auto') as batch:
for name in removable:
batch.drop_column(name)
+1
View File
@@ -87,6 +87,7 @@ _ALEMBIC_TENANT_TABLES = {
'monitoring_errors',
'monitoring_embedding_calls',
'monitoring_feedback',
'workspace_operation_logs',
'langbot_vectors',
'directory_projection_states',
'directory_projection_inbox',
@@ -72,6 +72,9 @@ TENANT_TABLE_COLUMNS: dict[str, str] = {
'monitoring_errors': 'workspace_uuid',
'monitoring_embedding_calls': 'workspace_uuid',
'monitoring_feedback': 'workspace_uuid',
# Administrative traceability for member and resource operations. Owned by
# the Workspace and therefore protected by the same RLS contract.
'workspace_operation_logs': 'workspace_uuid',
# Created by 0013 rather than ORM metadata; it is still part of the same
# business-database RLS contract and permits no discovery policies.
'langbot_vectors': 'workspace_uuid',