Skip to content

Organisation

Company structure facades: company, departments, teams, and role versions.

Enums

enums

Organization domain enumerations.

CompanyType

Bases: StrEnum

Pre-defined company template types.

DepartmentName

Bases: StrEnum

Standard department names within the organization.

Models

models

Organization-layer domain models.

Holds the input models for company, department, and agent mutations so the organization service layer can validate the same input shapes without importing from the API layer. The synthorg.api.dto_org module re-exports these for HTTP controllers.

UpdateCompanyRequest pydantic-model

Bases: BaseModel

Partial update for company-level settings.

Lives in the organization domain layer so the service can validate input without importing from synthorg.api.dto_org.

Config:

  • frozen: True
  • allow_inf_nan: False
  • extra: forbid

Fields:

company_name pydantic-field 

company_name = None

Display name of the company.

autonomy_level pydantic-field 

autonomy_level = None

Org-wide autonomy level (full, semi, supervised, locked).

budget_monthly pydantic-field 

budget_monthly = None

Monthly budget cap for the company in the operator's configured currency; set to 0 to disable enforcement.

communication_pattern pydantic-field 

communication_pattern = None

Communication strategy or pattern identifier.

CreateDepartmentRequest pydantic-model

Bases: BaseModel

Request body for creating a new department.

Config:

  • frozen: True
  • allow_inf_nan: False
  • extra: forbid

Fields:

UpdateDepartmentRequest pydantic-model

Bases: BaseModel

Partial update for an existing department.

Config:

  • frozen: True
  • allow_inf_nan: False
  • extra: forbid

Fields:

  • head (NotBlankStr | None)
  • budget_percent (float | None)
  • autonomy_level (AutonomyLevel | None)
  • teams (tuple[Team, ...] | None)
  • ceremony_policy (dict[str, object] | None)

Validators:

  • _validate_ceremony_policy → ceremony_policy

ReorderDepartmentsRequest pydantic-model

Bases: BaseModel

Reorder departments -- must be an exact permutation.

Config:

  • frozen: True
  • allow_inf_nan: False
  • extra: forbid

Fields:

CreateAgentOrgRequest pydantic-model

Bases: BaseModel

Request body for creating a new agent in the org config.

Config:

  • frozen: True
  • allow_inf_nan: False
  • extra: forbid

Fields:

Validators:

  • _validate_model_pair

UpdateAgentOrgRequest pydantic-model

Bases: BaseModel

Partial update for an existing agent in the org config.

Config:

  • frozen: True
  • allow_inf_nan: False
  • extra: forbid

Fields:

Validators:

  • _validate_model_pair

ReorderAgentsRequest pydantic-model

Bases: BaseModel

Reorder agents within a department -- must be an exact permutation.

Config:

  • frozen: True
  • allow_inf_nan: False
  • extra: forbid

Fields:

Services

services

Organization facades for the MCP handler layer.

Company read/versions, department CRUD + health, team CRUD, and role-version history. Writes route through org_mutation_service where it already owns the flow; other paths use in-memory stores until durable repositories land.

The file-level EM101 suppression is intentional: every capability gap in this module raises :class:CapabilityNotSupportedError from a short identifier and operator-readable reason, both string literals by design so capability telemetry (logged via :data:MCP_HANDLER_CAPABILITY_GAP) has a stable, grep-able message. Hoisting them to msg = ... locals would obscure the one-line intent with no runtime benefit.

UnsetType

Sentinel type for "field not provided" distinct from None.

Used by update operations where None is a legitimate value the caller may want to persist (e.g. clearing a department_id) and must be distinguished from "leave this field unchanged".

CompanyReadService 

CompanyReadService(*, org_mutation)

Read + light mutation facade over the company/org surface.

Source code in src/synthorg/organization/services.py 
81
82
def __init__(self, *, org_mutation: OrgMutationService) -> None:
    self._org = cast("Any", org_mutation)

get_company async 

get_company()

Return the company snapshot or raise if the capability is missing.

Raises:

Type Description
CapabilityNotSupportedError

When the wired OrgMutationService does not expose get_company.
Source code in src/synthorg/organization/services.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
async def get_company(self) -> object:
    """Return the company snapshot or raise if the capability is missing.

    Raises:
        CapabilityNotSupportedError: When the wired
            ``OrgMutationService`` does not expose ``get_company``.
    """
    fn = getattr(self._org, "get_company", None)
    if callable(fn):
        return await fn()
    raise CapabilityNotSupportedError(
        "company_get",
        "OrgMutationService does not expose get_company",
    )

update_company async 

update_company(*, payload, actor_id)

Apply a partial company-settings update.

The payload is validated against :class:UpdateCompanyRequest here so unknown keys are rejected before reaching the mutation service. The call then matches the target signature update_company(data, *, saved_by=...) -- actor_id is recorded as saved_by in the persisted snapshot.

Returns:

Type Description
object

The updated company snapshot returned by the mutation service.

Raises:

Type Description
CapabilityNotSupportedError

When OrgMutationService does not expose update_company.
Source code in src/synthorg/organization/services.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
async def update_company(
    self,
    *,
    payload: Mapping[str, object],
    actor_id: NotBlankStr,
) -> object:
    """Apply a partial company-settings update.

    The payload is validated against :class:`UpdateCompanyRequest`
    here so unknown keys are rejected before reaching the mutation
    service.  The call then matches the target signature
    ``update_company(data, *, saved_by=...)`` -- ``actor_id`` is
    recorded as ``saved_by`` in the persisted snapshot.

    Returns:
        The updated company snapshot returned by the mutation service.

    Raises:
        CapabilityNotSupportedError: When ``OrgMutationService`` does
            not expose ``update_company``.
    """
    fn = getattr(self._org, "update_company", None)
    if not callable(fn):
        raise CapabilityNotSupportedError(
            "company_update",
            "OrgMutationService does not expose update_company",
        )
    data = UpdateCompanyRequest.model_validate(dict(payload))
    result = await fn(data, saved_by=actor_id)
    logger.info(COMPANY_UPDATED_VIA_MCP, actor_id=actor_id)
    return result

list_departments async 

list_departments()

Return the company's departments.

Raises:

Type Description
CapabilityNotSupportedError

When OrgMutationService does not expose list_departments.
Source code in src/synthorg/organization/services.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
async def list_departments(self) -> Sequence[object]:
    """Return the company's departments.

    Raises:
        CapabilityNotSupportedError: When ``OrgMutationService`` does
            not expose ``list_departments``.
    """
    fn = getattr(self._org, "list_departments", None)
    if not callable(fn):
        raise CapabilityNotSupportedError(
            "company_list_departments",
            "OrgMutationService does not expose list_departments",
        )
    return tuple(await fn())

reorder_departments async 

reorder_departments(*, department_ids, actor_id)

Apply a new department ordering, auditing the change.

Raises:

Type Description
CapabilityNotSupportedError

When OrgMutationService does not expose reorder_departments.
Source code in src/synthorg/organization/services.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
async def reorder_departments(
    self,
    *,
    department_ids: Sequence[str],
    actor_id: NotBlankStr,
) -> None:
    """Apply a new department ordering, auditing the change.

    Raises:
        CapabilityNotSupportedError: When ``OrgMutationService`` does
            not expose ``reorder_departments``.
    """
    fn = getattr(self._org, "reorder_departments", None)
    if not callable(fn):
        raise CapabilityNotSupportedError(
            "company_reorder_departments",
            "OrgMutationService does not expose reorder_departments",
        )
    await fn(department_ids=tuple(department_ids), actor=actor_id)
    logger.info(
        DEPARTMENTS_REORDERED_VIA_MCP,
        actor_id=actor_id,
        count=len(department_ids),
    )

list_versions async 

list_versions()

Return all company-snapshot versions.

Raises:

Type Description
CapabilityNotSupportedError

When OrgMutationService does not expose list_company_versions.
Source code in src/synthorg/organization/services.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
async def list_versions(self) -> Sequence[object]:
    """Return all company-snapshot versions.

    Raises:
        CapabilityNotSupportedError: When ``OrgMutationService`` does
            not expose ``list_company_versions``.
    """
    fn = getattr(self._org, "list_company_versions", None)
    if not callable(fn):
        raise CapabilityNotSupportedError(
            "company_list_versions",
            "OrgMutationService does not expose list_company_versions",
        )
    return tuple(await fn())

get_version async 

get_version(version_id)

Fetch a single company-snapshot version by id.

Returns:

Type Description
object | None

The matching company-snapshot version, or None when no
object | None

version has that id.

Raises:

Type Description
CapabilityNotSupportedError

When OrgMutationService does not expose get_company_version.
Source code in src/synthorg/organization/services.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
async def get_version(self, version_id: NotBlankStr) -> object | None:
    """Fetch a single company-snapshot version by id.

    Returns:
        The matching company-snapshot version, or ``None`` when no
        version has that id.

    Raises:
        CapabilityNotSupportedError: When ``OrgMutationService`` does
            not expose ``get_company_version``.
    """
    fn = getattr(self._org, "get_company_version", None)
    if not callable(fn):
        raise CapabilityNotSupportedError(
            "company_get_version",
            "OrgMutationService does not expose get_company_version",
        )
    return cast("object | None", await fn(version_id))

DepartmentService 

DepartmentService()

Department CRUD + health.

Mutations are serialised through a single :class:asyncio.Lock so concurrent MCP handler calls cannot race on the in-memory dict (check-then-act in :meth:update_department and :meth:delete_department).

Source code in src/synthorg/organization/services.py 
245
246
247
def __init__(self) -> None:
    self._departments: dict[UUID, _DepartmentRecord] = {}
    self._lock = asyncio.Lock()

list_departments async 

list_departments(*, offset=0, limit=None)

Return paginated departments newest-first plus unfiltered total.

Parameters:

Name Type Description Default
offset int

Non-negative page offset.

 0
limit int | None

Optional positive page size; None returns every department from offset onwards.

 None

Raises:

Type Description
ValueError

If offset is negative, or limit is provided and non-positive.
Source code in src/synthorg/organization/services.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
async def list_departments(
    self,
    *,
    offset: int = 0,
    limit: int | None = None,
) -> tuple[tuple[_DepartmentRecord, ...], int]:
    """Return paginated departments newest-first plus unfiltered total.

    Args:
        offset: Non-negative page offset.
        limit: Optional positive page size; ``None`` returns every
            department from ``offset`` onwards.

    Raises:
        ValueError: If ``offset`` is negative, or ``limit`` is
            provided and non-positive.
    """
    if offset < 0:
        msg = f"offset must be >= 0, got {offset}"
        raise ValueError(msg)
    if limit is not None and limit < 1:
        msg = f"limit must be >= 1 when provided, got {limit}"
        raise ValueError(msg)
    async with self._lock:
        snapshot = tuple(copy.deepcopy(d) for d in self._departments.values())
    ordered = tuple(
        sorted(snapshot, key=lambda d: d.created_at, reverse=True),
    )
    total = len(ordered)
    end = total if limit is None else offset + limit
    return ordered[offset:end], total

get_department async 

get_department(department_id)

Fetch a single department by UUID or None if not found.

Returns:

Type Description
_DepartmentRecord | None

A deep copy of the stored department, or None when the id
_DepartmentRecord | None

is malformed or no such department exists.
Source code in src/synthorg/organization/services.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
async def get_department(
    self,
    department_id: NotBlankStr,
) -> _DepartmentRecord | None:
    """Fetch a single department by UUID or ``None`` if not found.

    Returns:
        A deep copy of the stored department, or ``None`` when the id
        is malformed or no such department exists.
    """
    try:
        key = UUID(department_id)
    except ValueError:
        return None
    async with self._lock:
        record = self._departments.get(key)
        return copy.deepcopy(record) if record is not None else None

create_department async 

create_department(*, name, description, actor_id)

Create a department, auditing the event on success.

Returns:

Type Description
_DepartmentRecord

A deep copy of the newly created department record.
Source code in src/synthorg/organization/services.py 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
async def create_department(
    self,
    *,
    name: NotBlankStr,
    description: NotBlankStr,
    actor_id: NotBlankStr,
) -> _DepartmentRecord:
    """Create a department, auditing the event on success.

    Returns:
        A deep copy of the newly created department record.
    """
    record = _DepartmentRecord(
        id=uuid4(),
        name=name,
        description=description,
        created_at=datetime.now(UTC),
    )
    async with self._lock:
        self._departments[record.id] = record
    logger.info(
        DEPARTMENT_CREATED_VIA_MCP,
        department_id=str(record.id),
        actor_id=actor_id,
    )
    return copy.deepcopy(record)

update_department async 

update_department(*, department_id, actor_id, name=None, description=None)

Patch a department's name / description in place.

Returns:

Type Description
_DepartmentRecord | None

A deep copy of the updated department, or None when the id
_DepartmentRecord | None

is malformed or no such department exists.
Source code in src/synthorg/organization/services.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
async def update_department(
    self,
    *,
    department_id: NotBlankStr,
    actor_id: NotBlankStr,
    name: NotBlankStr | None = None,
    description: NotBlankStr | None = None,
) -> _DepartmentRecord | None:
    """Patch a department's ``name`` / ``description`` in place.

    Returns:
        A deep copy of the updated department, or ``None`` when the id
        is malformed or no such department exists.
    """
    try:
        key = UUID(department_id)
    except ValueError:
        return None
    async with self._lock:
        record = self._departments.get(key)
        if record is None:
            return None
        if name is not None:
            record.name = name
        if description is not None:
            record.description = description
        record.updated_at = datetime.now(UTC)
        returned = copy.deepcopy(record)
    logger.info(
        DEPARTMENT_UPDATED_VIA_MCP,
        department_id=department_id,
        actor_id=actor_id,
    )
    return returned

delete_department async 

delete_department(*, department_id, actor_id, reason)

Remove a department; emit the audit event only on real removal.

Returns:

Type Description
bool

True when a department was removed, False when the id
bool

is malformed or no such department exists.
Source code in src/synthorg/organization/services.py 
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
async def delete_department(
    self,
    *,
    department_id: NotBlankStr,
    actor_id: NotBlankStr,
    reason: NotBlankStr,
) -> bool:
    """Remove a department; emit the audit event only on real removal.

    Returns:
        ``True`` when a department was removed, ``False`` when the id
        is malformed or no such department exists.
    """
    try:
        key = UUID(department_id)
    except ValueError:
        return False
    async with self._lock:
        removed = self._departments.pop(key, None) is not None
    if removed:
        logger.info(
            DEPARTMENT_DELETED_VIA_MCP,
            department_id=department_id,
            actor_id=actor_id,
            reason=reason,
            removed=removed,
        )
    return removed

get_health async 

get_health(department_id)

Return a summary health payload for the department.

Source code in src/synthorg/organization/services.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
async def get_health(
    self,
    department_id: NotBlankStr,
) -> Mapping[str, object]:
    """Return a summary health payload for the department."""
    record = await self.get_department(department_id)
    if record is None:
        return {"status": "unknown", "reason": "not_found"}
    return {
        "status": "healthy",
        "department_id": str(record.id),
        "last_update": record.updated_at.isoformat(),
    }

TeamService 

TeamService()

Team CRUD.

Mutations are serialised through a single :class:asyncio.Lock so concurrent MCP handler calls cannot race on the in-memory dict.

Source code in src/synthorg/organization/services.py 
443
444
445
def __init__(self) -> None:
    self._teams: dict[UUID, _TeamRecord] = {}
    self._lock = asyncio.Lock()

list_teams async 

list_teams(*, offset=0, limit=None)

Return paginated teams newest-first plus unfiltered total.

Parameters:

Name Type Description Default
offset int

Non-negative page offset.

 0
limit int | None

Optional positive page size; None returns every team from offset onwards.

 None

Raises:

Type Description
ValueError

If offset is negative, or limit is provided and non-positive.
Source code in src/synthorg/organization/services.py 
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
async def list_teams(
    self,
    *,
    offset: int = 0,
    limit: int | None = None,
) -> tuple[tuple[_TeamRecord, ...], int]:
    """Return paginated teams newest-first plus unfiltered total.

    Args:
        offset: Non-negative page offset.
        limit: Optional positive page size; ``None`` returns every
            team from ``offset`` onwards.

    Raises:
        ValueError: If ``offset`` is negative, or ``limit`` is
            provided and non-positive.
    """
    if offset < 0:
        msg = f"offset must be >= 0, got {offset}"
        raise ValueError(msg)
    if limit is not None and limit < 1:
        msg = f"limit must be >= 1 when provided, got {limit}"
        raise ValueError(msg)
    async with self._lock:
        snapshot = tuple(copy.deepcopy(t) for t in self._teams.values())
    ordered = tuple(
        sorted(snapshot, key=lambda t: t.created_at, reverse=True),
    )
    total = len(ordered)
    end = total if limit is None else offset + limit
    return ordered[offset:end], total

get_team async 

get_team(team_id)

Fetch a single team by UUID or None if not found.

Returns:

Type Description
_TeamRecord | None

A deep copy of the stored team, or None when the id is
_TeamRecord | None

malformed or no such team exists.
Source code in src/synthorg/organization/services.py 
479
480
481
482
483
484
485
486
487
488
489
490
491
492
async def get_team(self, team_id: NotBlankStr) -> _TeamRecord | None:
    """Fetch a single team by UUID or ``None`` if not found.

    Returns:
        A deep copy of the stored team, or ``None`` when the id is
        malformed or no such team exists.
    """
    try:
        key = UUID(team_id)
    except ValueError:
        return None
    async with self._lock:
        record = self._teams.get(key)
        return copy.deepcopy(record) if record is not None else None

create_team async 

create_team(*, name, actor_id, department_id=None)

Create a team, auditing the event on success.

Returns:

Type Description
_TeamRecord

A deep copy of the newly created team record.
Source code in src/synthorg/organization/services.py 
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
async def create_team(
    self,
    *,
    name: NotBlankStr,
    actor_id: NotBlankStr,
    department_id: NotBlankStr | None = None,
) -> _TeamRecord:
    """Create a team, auditing the event on success.

    Returns:
        A deep copy of the newly created team record.
    """
    record = _TeamRecord(
        id=uuid4(),
        name=name,
        department_id=department_id,
        created_at=datetime.now(UTC),
    )
    async with self._lock:
        self._teams[record.id] = record
    logger.info(
        TEAM_CREATED_VIA_MCP,
        team_id=str(record.id),
        actor_id=actor_id,
    )
    return copy.deepcopy(record)

update_team async 

update_team(*, team_id, actor_id, name=None, department_id=UNSET)

Update a team; department_id=None clears the field.

The default department_id=UNSET sentinel means "leave unchanged"; pass department_id=None explicitly to clear a team's department assignment.

Returns:

Type Description
_TeamRecord | None

A deep copy of the updated team, or None when the id is
_TeamRecord | None

malformed or no such team exists.
Source code in src/synthorg/organization/services.py 
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
async def update_team(
    self,
    *,
    team_id: NotBlankStr,
    actor_id: NotBlankStr,
    name: NotBlankStr | None = None,
    department_id: NotBlankStr | None | UnsetType = UNSET,
) -> _TeamRecord | None:
    """Update a team; ``department_id=None`` clears the field.

    The default ``department_id=UNSET`` sentinel means "leave
    unchanged"; pass ``department_id=None`` explicitly to clear a
    team's department assignment.

    Returns:
        A deep copy of the updated team, or ``None`` when the id is
        malformed or no such team exists.
    """
    try:
        key = UUID(team_id)
    except ValueError:
        return None
    async with self._lock:
        record = self._teams.get(key)
        if record is None:
            return None
        if name is not None:
            record.name = name
        if not isinstance(department_id, UnsetType):
            record.department_id = department_id
        record.updated_at = datetime.now(UTC)
        returned = copy.deepcopy(record)
    logger.info(
        TEAM_UPDATED_VIA_MCP,
        team_id=team_id,
        actor_id=actor_id,
    )
    return returned

delete_team async 

delete_team(*, team_id, actor_id, reason)

Remove a team; emit the audit event only on real removal.

Returns:

Type Description
bool

True when a team was removed, False when the id is
bool

malformed or no such team exists.
Source code in src/synthorg/organization/services.py 
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
async def delete_team(
    self,
    *,
    team_id: NotBlankStr,
    actor_id: NotBlankStr,
    reason: NotBlankStr,
) -> bool:
    """Remove a team; emit the audit event only on real removal.

    Returns:
        ``True`` when a team was removed, ``False`` when the id is
        malformed or no such team exists.
    """
    try:
        key = UUID(team_id)
    except ValueError:
        return False
    async with self._lock:
        removed = self._teams.pop(key, None) is not None
    if removed:
        logger.info(
            TEAM_DELETED_VIA_MCP,
            team_id=team_id,
            actor_id=actor_id,
            reason=reason,
            removed=removed,
        )
    return removed

RoleVersionService 

RoleVersionService(*, org_mutation=None)

Read facade for the role-version snapshot history.

Source code in src/synthorg/organization/services.py 
596
597
598
599
600
601
def __init__(
    self,
    *,
    org_mutation: OrgMutationService | None = None,
) -> None:
    self._org = cast("Any", org_mutation) if org_mutation is not None else None

list_versions async 

list_versions(*, role_name=None)

Return role-snapshot versions, optionally filtered by role name.

Raises:

Type Description
CapabilityNotSupportedError

When no OrgMutationService is wired, or the wired one does not expose list_role_versions.
Source code in src/synthorg/organization/services.py 
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
async def list_versions(
    self,
    *,
    role_name: NotBlankStr | None = None,
) -> Sequence[object]:
    """Return role-snapshot versions, optionally filtered by role name.

    Raises:
        CapabilityNotSupportedError: When no ``OrgMutationService`` is
            wired, or the wired one does not expose
            ``list_role_versions``.
    """
    if self._org is None:
        raise CapabilityNotSupportedError(
            "role_versions_list",
            "OrgMutationService not wired on app_state",
        )
    fn = getattr(self._org, "list_role_versions", None)
    if not callable(fn):
        raise CapabilityNotSupportedError(
            "role_versions_list",
            "OrgMutationService does not expose list_role_versions",
        )
    return tuple(await fn(role_name=role_name))

get_version async 

get_version(version_id)

Fetch a single role-snapshot version by id.

Returns:

Type Description
object | None

The matching role-snapshot version, or None when no
object | None

version has that id.

Raises:

Type Description
CapabilityNotSupportedError

When no OrgMutationService is wired, or the wired one does not expose get_role_version.
Source code in src/synthorg/organization/services.py 
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
async def get_version(
    self,
    version_id: NotBlankStr,
) -> object | None:
    """Fetch a single role-snapshot version by id.

    Returns:
        The matching role-snapshot version, or ``None`` when no
        version has that id.

    Raises:
        CapabilityNotSupportedError: When no ``OrgMutationService`` is
            wired, or the wired one does not expose
            ``get_role_version``.
    """
    if self._org is None:
        raise CapabilityNotSupportedError(
            "role_versions_get",
            "OrgMutationService not wired on app_state",
        )
    fn = getattr(self._org, "get_role_version", None)
    if not callable(fn):
        raise CapabilityNotSupportedError(
            "role_versions_get",
            "OrgMutationService does not expose get_role_version",
        )
    return cast("object | None", await fn(version_id))