diff --git a/src/langbot/pkg/api/application/api.py b/src/langbot/pkg/api/application/api.py index a46116fce..8f6e42364 100644 --- a/src/langbot/pkg/api/application/api.py +++ b/src/langbot/pkg/api/application/api.py @@ -265,12 +265,14 @@ class ApplicationAPI: resource_id = next(iter(identity.values())) detail = {'operation': spec.operation_id, 'version': spec.version} actor_name = None + dedupe_actor = None if context.principal.principal_type == PrincipalType.PROCESSOR: actor_name = ( f'{context.principal.processor_kind}:{context.principal.processor_uuid}' if context.principal.processor_kind else context.principal.processor_uuid ) + dedupe_actor = f'processor:{context.principal.processor_uuid}:{context.principal.run_id}' detail.update( { 'processor_kind': context.principal.processor_kind, @@ -297,6 +299,7 @@ class ApplicationAPI: status_code=status, detail=detail, duration_ms=int((time.monotonic() - observed_at) * 1000), + dedupe_actor=dedupe_actor, ) except Exception: # Audit failures must preserve the original operation result or error. diff --git a/src/langbot/pkg/api/application/processor_context.py b/src/langbot/pkg/api/application/processor_context.py index bb29312c2..52d73c94b 100644 --- a/src/langbot/pkg/api/application/processor_context.py +++ b/src/langbot/pkg/api/application/processor_context.py @@ -8,6 +8,8 @@ from collections.abc import Iterable from ..context import PrincipalContext, PrincipalType, RequestContext, WorkspaceContext +_VALID_PROCESSOR_KINDS: typing.Final = frozenset({'agent', 'pipeline', 'event_processor'}) + async def build_processor_request_context( ap: typing.Any, @@ -20,22 +22,20 @@ async def build_processor_request_context( trigger_actor: str | None = None, ) -> RequestContext: """Build a trusted RequestContext for internal processor invocations.""" - workspace_service = getattr(ap, 'workspace_service', None) - if workspace_service is None and hasattr(ap, 'ap'): - workspace_service = getattr(ap.ap, 'workspace_service', None) - if workspace_service is None: - raise RuntimeError('Workspace service is required to resolve execution binding') + if not isinstance(workspace_uuid, str) or not workspace_uuid.strip(): + raise ValueError('workspace_uuid is required') + if processor_kind not in _VALID_PROCESSOR_KINDS: + raise ValueError(f'Invalid processor_kind {processor_kind!r}; must be one of {sorted(_VALID_PROCESSOR_KINDS)}') + if not isinstance(processor_uuid, str) or not processor_uuid.strip(): + raise ValueError('processor_uuid is required') + if not isinstance(run_id, str) or not run_id.strip(): + raise ValueError('run_id is required') - binding = await workspace_service.get_execution_binding(workspace_uuid) - - app_api = getattr(ap, 'application_api', None) - if app_api is None and hasattr(ap, '_operations'): - app_api = ap - if app_api is None and hasattr(ap, 'ap'): - app_api = getattr(ap.ap, 'application_api', None) + binding = await ap.workspace_service.get_execution_binding(workspace_uuid) allowed_ids = frozenset(operation_ids) permissions: set[str] = set() + app_api = getattr(ap, 'application_api', None) if app_api is not None: operations = getattr(app_api, '_operations', {}) for op_id in allowed_ids: diff --git a/src/langbot/pkg/operation_trace/service.py b/src/langbot/pkg/operation_trace/service.py index 98e335e67..dbd281e8f 100644 --- a/src/langbot/pkg/operation_trace/service.py +++ b/src/langbot/pkg/operation_trace/service.py @@ -1127,6 +1127,7 @@ def dedupe_material( action: str, route: str | None, resource_id: str | None, + dedupe_actor: str | None = None, ) -> str: """Return the raw material the dedupe key hashes. @@ -1134,9 +1135,10 @@ def dedupe_material( so enabling tracing never runs a hash on the request path. """ + actor_key = dedupe_actor if dedupe_actor is not None else str(actor_account_uuid or '') return '|'.join( [ - str(actor_account_uuid or ''), + actor_key, str(action or ''), str(route or ''), str(resource_id or ''), @@ -1156,6 +1158,7 @@ def compute_dedupe_key( action: str, route: str | None, resource_id: str | None, + dedupe_actor: str | None = None, ) -> str: """Build the key that collapses repeated identical observations.""" @@ -1165,6 +1168,7 @@ def compute_dedupe_key( action=action, route=route, resource_id=resource_id, + dedupe_actor=dedupe_actor, ) ) @@ -1578,6 +1582,7 @@ class WorkspaceSettingsService: client_ip: str | None = None, user_agent: str | None = None, duration_ms: int = 0, + dedupe_actor: str | None = None, ) -> bool: """Queue one operation for traceability. Returns whether it was queued. @@ -1630,6 +1635,7 @@ class WorkspaceSettingsService: action=rule.action, route=record_fields['route'], resource_id=record_fields['resource_id'], + dedupe_actor=dedupe_actor, ) return self._enqueue(record_fields)