mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-23 08:55:44 +08:00
refactor(operation-trace): isolate subsystem, fix install identity, merge migrations
- Move the operation-traceability service and its HTTP routes into a standalone pkg/operation_trace package; the Core controller package no longer depends on it (routes stay auto-discovered by the controller package scan). - Gate the hot path on one module-level boolean exposed as ap.operation_trace_active: while no Workspace has tracing enabled the route wrapper performs zero trace work and zero database round trips. The gate opens when a Workspace enables tracing and is primed once at startup for already-enabled Workspaces. - Record the installed extension identity for GitHub / marketplace / local / upload installs (owner/repo, author/name, filename) so the trace no longer shows a nameless "a plugin was installed" event and the card names the extension. - Merge migrations 0032 + 0033 into a single 0032_operation_traceability revision carrying the full table shape and an index set that matches the ORM. Tests: unit 5165 passed / integration 358 passed; ruff check + format clean.
This commit is contained in:
@@ -363,6 +363,13 @@ class RouterGroup(abc.ABC):
|
||||
database, so a disabled Workspace pays nothing at steady state.
|
||||
"""
|
||||
|
||||
# Cheap global gate: a single attribute read on the application object.
|
||||
# While no Workspace has enabled traceability this is ``False`` and the
|
||||
# isolated subsystem stays entirely off the hot request path, with no
|
||||
# import, service lookup, JSON parse or database round trip.
|
||||
if not getattr(self.ap, 'operation_trace_active', False):
|
||||
return
|
||||
|
||||
service = getattr(self.ap, 'workspace_settings_service', None)
|
||||
if service is None:
|
||||
return
|
||||
@@ -394,16 +401,15 @@ class RouterGroup(abc.ABC):
|
||||
# touching each handler. The body is read defensively: an audit path
|
||||
# must never fail (or block) because a request carries no JSON.
|
||||
try:
|
||||
# Imported lazily so the base controller never participates in an
|
||||
# import cycle with the service layer at module load time.
|
||||
from ..service import settings as settings_service
|
||||
|
||||
# Identity resolution lives in the isolated traceability service
|
||||
# and is reached through the handle, so the Core controller never
|
||||
# imports that subsystem.
|
||||
request_body: dict[str, typing.Any] | None = None
|
||||
if str(quart.request.method or '').upper() not in ('GET', 'HEAD', 'OPTIONS'):
|
||||
raw_body = await quart.request.get_json(silent=True)
|
||||
if isinstance(raw_body, dict):
|
||||
request_body = raw_body
|
||||
identity = settings_service.resolve_resource_identity(kwargs, request_body)
|
||||
identity = service.resolve_resource_identity(kwargs, request_body)
|
||||
if identity:
|
||||
resource_id = identity
|
||||
except Exception: # pragma: no cover - identity resolution is best effort
|
||||
|
||||
@@ -2,7 +2,7 @@ import quart
|
||||
|
||||
from ....authz import Permission, has_permission
|
||||
from ....context import RequestContext
|
||||
from ....service import settings as settings_service
|
||||
from ......operation_trace import service as settings_service
|
||||
from ... import group
|
||||
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@ import quart
|
||||
|
||||
from ....authz import Permission, has_permission
|
||||
from ....context import RequestContext
|
||||
from ....service import settings as settings_service
|
||||
from ......operation_trace import service as settings_service
|
||||
from ....service.secrets import redact_secrets
|
||||
from ... import group
|
||||
from ......pipeline.extension_preferences import (
|
||||
|
||||
@@ -18,7 +18,7 @@ from .....core import taskmgr
|
||||
from .....entity.persistence import plugin as persistence_plugin
|
||||
from ...authz import Permission
|
||||
from ...context import ExecutionContext, RequestContext
|
||||
from ...service import settings as settings_service
|
||||
from .....operation_trace import service as settings_service
|
||||
from .. import group
|
||||
from .....workspace.errors import WorkspaceNotFoundError
|
||||
from .....plugin.github import validate_github_plugin_install_info
|
||||
@@ -872,6 +872,11 @@ class PluginsRouterGroup(group.RouterGroup):
|
||||
owner = install_info['owner']
|
||||
repo = install_info['repo']
|
||||
release_tag = install_info['release_tag']
|
||||
# Name the installed extension in the trace. GitHub installs carry
|
||||
# owner/repo in the body, which the audit identity resolver can also
|
||||
# read, but publishing it here keeps the trace correct regardless of
|
||||
# how the request was encoded.
|
||||
quart.g.operation_log_resource_id = f'{owner}/{repo}'
|
||||
|
||||
execution_context = await self.ap.plugin_connector.require_workspace_context(request_context)
|
||||
|
||||
@@ -912,6 +917,7 @@ class PluginsRouterGroup(group.RouterGroup):
|
||||
|
||||
plugin_author = data.get('plugin_author', '')
|
||||
plugin_name = data.get('plugin_name', '')
|
||||
quart.g.operation_log_resource_id = f'{plugin_author}/{plugin_name}'
|
||||
execution_context = await self.ap.plugin_connector.require_workspace_context(request_context)
|
||||
|
||||
ctx = taskmgr.TaskContext.new()
|
||||
@@ -953,6 +959,9 @@ class PluginsRouterGroup(group.RouterGroup):
|
||||
file_bytes = file.read()
|
||||
form = await quart.request.form
|
||||
administrator_force = form.get('administrator_force', '').strip().lower() == 'true'
|
||||
# A local upload is multipart with no JSON body, so the archive
|
||||
# filename is the only identity available for the trace.
|
||||
quart.g.operation_log_resource_id = file.filename or 'local plugin'
|
||||
execution_context = await self.ap.plugin_connector.require_workspace_context(request_context)
|
||||
|
||||
data = {
|
||||
|
||||
@@ -3,7 +3,7 @@ import quart
|
||||
from langbot.pkg.provider.modelmgr import errors as provider_errors
|
||||
from ....authz import Permission, has_permission
|
||||
from ....context import RequestContext
|
||||
from ....service import settings as settings_service
|
||||
from ......operation_trace import service as settings_service
|
||||
from ... import group
|
||||
from .query import resolve_include_secret
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@ import quart
|
||||
|
||||
from ....authz import Permission, has_permission
|
||||
from ....context import RequestContext
|
||||
from ....service import settings as settings_service
|
||||
from ......operation_trace import service as settings_service
|
||||
from ... import group
|
||||
from .query import resolve_include_secret
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ from urllib.parse import unquote
|
||||
|
||||
from ....authz import Permission
|
||||
from ....context import RequestContext
|
||||
from ....service import settings as settings_service
|
||||
from ......operation_trace import service as settings_service
|
||||
from ......provider.tools.loaders.mcp_policy import MCPStdioDisabledError
|
||||
from ... import group
|
||||
|
||||
|
||||
@@ -1,250 +0,0 @@
|
||||
"""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
|
||||
* ``GET /api/v1/settings/operation-logs/export`` download the filtered CSV
|
||||
|
||||
Operation records are append-only and there is no route to delete them on
|
||||
demand: retention is the only deletion path, and it runs automatically when
|
||||
the governance settings change and from the maintenance loop.
|
||||
|
||||
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')),
|
||||
# Optional verification filter: lets the panel turn the three
|
||||
# tamper counters into a drill-down instead of a dead badge.
|
||||
integrity=args.get('integrity') or None,
|
||||
)
|
||||
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
|
||||
@@ -7,7 +7,7 @@ from langbot_plugin.box.errors import BoxError
|
||||
|
||||
from ...authz import Permission
|
||||
from ...context import RequestContext
|
||||
from ...service import settings as settings_service
|
||||
from .....operation_trace import service as settings_service
|
||||
from .. import group
|
||||
|
||||
|
||||
@@ -185,6 +185,9 @@ class SkillsRouterGroup(group.RouterGroup):
|
||||
if not asset_url.endswith('skill.md') and not data.get('release_tag'):
|
||||
return self.http_status(400, -1, 'Missing required field: release_tag')
|
||||
|
||||
# Name the installed skill in the trace (owner/repo from the body).
|
||||
quart.g.operation_log_resource_id = f'{data.get("owner", "")}/{data.get("repo", "")}'
|
||||
|
||||
try:
|
||||
skill = await self.ap.skill_service.install_from_github(request_context, data)
|
||||
return self.success(data={'skills': skill})
|
||||
@@ -228,6 +231,9 @@ class SkillsRouterGroup(group.RouterGroup):
|
||||
if file is None:
|
||||
return self.http_status(400, -1, 'file is required')
|
||||
form = await quart.request.form
|
||||
# A zip upload is multipart with no JSON body, so the filename is
|
||||
# the only identity available for the trace.
|
||||
quart.g.operation_log_resource_id = file.filename or 'skill'
|
||||
|
||||
try:
|
||||
skill = await self.ap.skill_service.install_from_zip_upload(
|
||||
|
||||
@@ -12,7 +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 .....operation_trace import service as settings_service
|
||||
from .. import group
|
||||
|
||||
|
||||
|
||||
@@ -20,6 +20,12 @@ from .groups import knowledge as groups_knowledge
|
||||
from .groups import resources as groups_resources
|
||||
from ...mcp.mount import MCPMount
|
||||
|
||||
# The traceability subsystem is self-contained and auto-registers its routes
|
||||
# through ``group.group_class``. Importing it here (instead of from within the
|
||||
# ``groups`` package) keeps the feature decoupled: the Core controller package
|
||||
# contains no code that depends on it, only this single bootstrap import.
|
||||
from ....operation_trace import routes as _operation_trace_routes # noqa: F401
|
||||
|
||||
importutil.import_modules_in_pkg(groups)
|
||||
importutil.import_modules_in_pkg(groups_provider)
|
||||
importutil.import_modules_in_pkg(groups_platform)
|
||||
|
||||
Reference in New Issue
Block a user