Skip to content

Collapse the actor-stamping factories onto the shared core #608

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
xmap merged 1 commit into from
Aug 4, 2026
+96 −409
Merged

Collapse the actor-stamping factories onto the shared core #608

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
162 changes: 28 additions & 134 deletions apps/api/src/cora/agent/_agent_update_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,20 +40,16 @@
event payload carries the canonical `<verb>_by` attribution half.
Mirrors `cora.federation._actor_update_handler.make_actor_update_handler`
byte-for-byte modulo the Agent-specific defaults; the body
duplicates `make_update_handler`'s flow because `principal_id` only
enters scope at handler-call time, not at factory-build time.

NOTE (superseded): `make_update_handler` now takes an optional
`actor_kwarg` and threads the principal itself, so this body no longer
HAS to be a copy. Collapsing this factory onto the shared core is a
recorded follow-up, deliberately not done in the commit that added the
parameter so that change stayed reviewable.
ONCE duplicated `make_update_handler`'s flow, because `principal_id` enters
scope at handler-call time rather than at factory-build time.

COLLAPSED 2026-08-03: `make_update_handler` takes an optional
`actor_kwarg` and threads the principal itself, so this factory is now a
thin wrapper like its non-stamping sibling and carries no body of its own.
"""

from collections.abc import Callable, Sequence
from datetime import datetime # noqa: TC003 (runtime-imported for clarity)
from typing import Any, Protocol
from uuid import UUID
from typing import Any

from cora.agent.aggregates.agent import (
AgentEvent,
Expand All @@ -63,13 +59,8 @@
to_payload,
)
from cora.agent.errors import UnauthorizedError
from cora.infrastructure.event_envelope import to_new_event
from cora.infrastructure.kernel import Kernel
from cora.infrastructure.logging import get_logger
from cora.infrastructure.ports import Deny
from cora.infrastructure.routing import NIL_SENTINEL_ID
from cora.infrastructure.update_handler import make_update_handler
from cora.shared.identity import ActorId

_STREAM_TYPE = "Agent"
_TARGET_ID_ATTR = "agent_id"
Expand Down Expand Up @@ -100,25 +91,6 @@ def make_agent_update_handler(
)


class _ActorUpdateHandler(Protocol):
"""Callable shape returned by `make_agent_actor_update_handler`.

Mirrors the cross-BC factory's `_UpdateHandler` shape so per-slice
`Handler` Protocols (which are narrower in `command`) keep
assigning without explicit casts.
"""

async def __call__(
self,
command: Any,
*,
principal_id: UUID,
correlation_id: UUID,
causation_id: UUID | None = None,
surface_id: UUID = NIL_SENTINEL_ID,
) -> None: ...


def make_agent_actor_update_handler(
deps: Kernel,
*,
Expand All @@ -127,106 +99,28 @@ def make_agent_actor_update_handler(
decide_fn: Callable[..., Sequence[AgentEvent]],
actor_kwarg: str,
extra_log_fields: Callable[[Any], dict[str, Any]] | None = None,
) -> _ActorUpdateHandler:
"""Build an actor-stamping update handler for one Agent slice.

`actor_kwarg` is the decider's `<verb>_by` parameter name; the
handler passes the envelope's `principal_id` (wrapped in `ActorId`)
under that name on every call. Used by fold-symmetry slices
(`suspend_agent`, `resume_agent`) whose events carry a folded
attribution half on the payload.
):
"""Build an update-style handler for one Agent slice, stamping the actor.

Delegates to the cross-BC factory. `actor_kwarg` names the decider keyword
that receives `ActorId(principal_id)`; the core threads it, so this wrapper
carries no body of its own.
"""
log = get_logger(log_prefix)

async def handler(
command: Any,
*,
principal_id: UUID,
correlation_id: UUID,
causation_id: UUID | None = None,
surface_id: UUID = NIL_SENTINEL_ID,
) -> None:
target_id: UUID = getattr(command, _TARGET_ID_ATTR)
extras: dict[str, Any] = extra_log_fields(command) if extra_log_fields is not None else {}

log.info(
f"{log_prefix}.start",
command_name=command_name,
**{_TARGET_ID_ATTR: str(target_id)},
**extras,
principal_id=str(principal_id),
correlation_id=str(correlation_id),
causation_id=str(causation_id) if causation_id is not None else None,
)

decision = await deps.authz.authorize(
principal_id=principal_id,
command_name=command_name,
conduit_id=NIL_SENTINEL_ID,
surface_id=surface_id,
)
if isinstance(decision, Deny):
log.info(
f"{log_prefix}.denied",
command_name=command_name,
**{_TARGET_ID_ATTR: str(target_id)},
**extras,
principal_id=str(principal_id),
correlation_id=str(correlation_id),
causation_id=str(causation_id) if causation_id is not None else None,
reason=decision.reason,
)
raise UnauthorizedError(decision.reason)

now: datetime = deps.clock.now()

stored, current_version = await deps.event_store.load(
stream_type=_STREAM_TYPE,
stream_id=target_id,
)
history: list[AgentEvent] = [from_stored(s) for s in stored]
state = fold(history)

domain_events = decide_fn(
state=state,
command=command,
now=now,
**{actor_kwarg: ActorId(principal_id)},
)

new_events = [
to_new_event(
event_type=event_type_name(event),
payload=to_payload(event),
occurred_at=event.occurred_at,
event_id=deps.id_generator.new_id(),
command_name=command_name,
correlation_id=correlation_id,
causation_id=causation_id,
principal_id=principal_id,
)
for event in domain_events
]
await deps.event_store.append(
stream_type=_STREAM_TYPE,
stream_id=target_id,
expected_version=current_version,
events=new_events,
)

log.info(
f"{log_prefix}.success",
command_name=command_name,
**{_TARGET_ID_ATTR: str(target_id)},
**extras,
principal_id=str(principal_id),
correlation_id=str(correlation_id),
causation_id=str(causation_id) if causation_id is not None else None,
event_count=len(new_events),
new_version=current_version + len(new_events),
)

return handler
return make_update_handler(
deps,
stream_type=_STREAM_TYPE,
target_id_attr=_TARGET_ID_ATTR,
from_stored=from_stored,
to_payload=to_payload,
event_type_name=event_type_name,
fold=fold,
unauthorized_error=UnauthorizedError,
command_name=command_name,
log_prefix=log_prefix,
decide_fn=decide_fn,
actor_kwarg=actor_kwarg,
extra_log_fields=extra_log_fields,
)


__all__ = ["make_agent_actor_update_handler", "make_agent_update_handler"]
162 changes: 28 additions & 134 deletions apps/api/src/cora/budget/_allocation_update_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,20 +26,16 @@

The actor variant mirrors `make_agent_actor_update_handler`
byte-for-byte modulo the Allocation-specific defaults; the body
duplicates `make_update_handler`'s flow because `principal_id` only
enters scope at handler-call time, not at factory-build time.

NOTE (superseded): `make_update_handler` now takes an optional
`actor_kwarg` and threads the principal itself, so this body no longer
HAS to be a copy. Collapsing this factory onto the shared core is a
recorded follow-up, deliberately not done in the commit that added the
parameter so that change stayed reviewable.
ONCE duplicated `make_update_handler`'s flow, because `principal_id` enters
scope at handler-call time rather than at factory-build time.

COLLAPSED 2026-08-03: `make_update_handler` takes an optional
`actor_kwarg` and threads the principal itself, so this factory is now a
thin wrapper like its non-stamping sibling and carries no body of its own.
"""

from collections.abc import Callable, Sequence
from datetime import datetime # noqa: TC003 (runtime-imported for clarity)
from typing import Any, Protocol
from uuid import UUID
from typing import Any

from cora.budget.aggregates.allocation import (
AllocationEvent,
Expand All @@ -49,13 +45,8 @@
to_payload,
)
from cora.budget.errors import UnauthorizedError
from cora.infrastructure.event_envelope import to_new_event
from cora.infrastructure.kernel import Kernel
from cora.infrastructure.logging import get_logger
from cora.infrastructure.ports import Deny
from cora.infrastructure.routing import NIL_SENTINEL_ID
from cora.infrastructure.update_handler import make_update_handler
from cora.shared.identity import ActorId

_STREAM_TYPE = "Allocation"
_TARGET_ID_ATTR = "allocation_id"
Expand Down Expand Up @@ -86,25 +77,6 @@ def make_allocation_update_handler(
)


class _ActorUpdateHandler(Protocol):
"""Callable shape returned by `make_allocation_actor_update_handler`.

Mirrors the cross-BC factory's `_UpdateHandler` shape so per-slice
`Handler` Protocols (which are narrower in `command`) keep
assigning without explicit casts.
"""

async def __call__(
self,
command: Any,
*,
principal_id: UUID,
correlation_id: UUID,
causation_id: UUID | None = None,
surface_id: UUID = NIL_SENTINEL_ID,
) -> None: ...


def make_allocation_actor_update_handler(
deps: Kernel,
*,
Expand All @@ -113,106 +85,28 @@ def make_allocation_actor_update_handler(
decide_fn: Callable[..., Sequence[AllocationEvent]],
actor_kwarg: str,
extra_log_fields: Callable[[Any], dict[str, Any]] | None = None,
) -> _ActorUpdateHandler:
"""Build an actor-stamping update handler for one Allocation slice.

`actor_kwarg` is the decider's `<verb>_by` parameter name; the
handler passes the envelope's `principal_id` (wrapped in `ActorId`)
under that name on every call. Used by fold-symmetry slices
(`activate_allocation`) whose events carry a folded attribution
half on the payload.
):
"""Build an update-style handler for one Allocation slice, stamping the actor.

Delegates to the cross-BC factory. `actor_kwarg` names the decider keyword
that receives `ActorId(principal_id)`; the core threads it, so this wrapper
carries no body of its own.
"""
log = get_logger(log_prefix)

async def handler(
command: Any,
*,
principal_id: UUID,
correlation_id: UUID,
causation_id: UUID | None = None,
surface_id: UUID = NIL_SENTINEL_ID,
) -> None:
target_id: UUID = getattr(command, _TARGET_ID_ATTR)
extras: dict[str, Any] = extra_log_fields(command) if extra_log_fields is not None else {}

log.info(
f"{log_prefix}.start",
command_name=command_name,
**{_TARGET_ID_ATTR: str(target_id)},
**extras,
principal_id=str(principal_id),
correlation_id=str(correlation_id),
causation_id=str(causation_id) if causation_id is not None else None,
)

decision = await deps.authz.authorize(
principal_id=principal_id,
command_name=command_name,
conduit_id=NIL_SENTINEL_ID,
surface_id=surface_id,
)
if isinstance(decision, Deny):
log.info(
f"{log_prefix}.denied",
command_name=command_name,
**{_TARGET_ID_ATTR: str(target_id)},
**extras,
principal_id=str(principal_id),
correlation_id=str(correlation_id),
causation_id=str(causation_id) if causation_id is not None else None,
reason=decision.reason,
)
raise UnauthorizedError(decision.reason)

now: datetime = deps.clock.now()

stored, current_version = await deps.event_store.load(
stream_type=_STREAM_TYPE,
stream_id=target_id,
)
history: list[AllocationEvent] = [from_stored(s) for s in stored]
state = fold(history)

domain_events = decide_fn(
state=state,
command=command,
now=now,
**{actor_kwarg: ActorId(principal_id)},
)

new_events = [
to_new_event(
event_type=event_type_name(event),
payload=to_payload(event),
occurred_at=event.occurred_at,
event_id=deps.id_generator.new_id(),
command_name=command_name,
correlation_id=correlation_id,
causation_id=causation_id,
principal_id=principal_id,
)
for event in domain_events
]
await deps.event_store.append(
stream_type=_STREAM_TYPE,
stream_id=target_id,
expected_version=current_version,
events=new_events,
)

log.info(
f"{log_prefix}.success",
command_name=command_name,
**{_TARGET_ID_ATTR: str(target_id)},
**extras,
principal_id=str(principal_id),
correlation_id=str(correlation_id),
causation_id=str(causation_id) if causation_id is not None else None,
event_count=len(new_events),
new_version=current_version + len(new_events),
)

return handler
return make_update_handler(
deps,
stream_type=_STREAM_TYPE,
target_id_attr=_TARGET_ID_ATTR,
from_stored=from_stored,
to_payload=to_payload,
event_type_name=event_type_name,
fold=fold,
unauthorized_error=UnauthorizedError,
command_name=command_name,
log_prefix=log_prefix,
decide_fn=decide_fn,
actor_kwarg=actor_kwarg,
extra_log_fields=extra_log_fields,
)


__all__ = ["make_allocation_actor_update_handler", "make_allocation_update_handler"]
Loading
Loading