Skip to content

identity_resolver

identity_resolver

Turn a container id into a task-aware ContainerIdentity.

Composes an injected ContainerInspector (container name + OCI annotations, produced by whichever runtime backend the caller plugged in) with a data-contract YAML lookup to produce notification-ready identities. Clearance carries no Python-import coupling to any orchestrator; the one that created the container publishes a path annotation pointing at its own metadata file, and the resolver reads .name from that file on each call.

.. rubric:: Annotation contract

Containers that want friendly task-name enrichment in clearance notifications set three OCI annotations at container creation:

ai.terok.project Project identifier (required for task-aware labelling).

ai.terok.task Task identifier within the project.

ai.terok.task_meta_path Absolute path to a YAML file. The resolver reads the name field and uses it as the display name. Missing / unreadable / malformed files fall back silently to the container id.

Rename during a container's lifetime is preserved: the annotation carries the path, not the name, and the orchestrator updates the YAML in place. Each resolver call reads the current value.

ANNOTATION_PROJECT = 'ai.terok.project' module-attribute

ANNOTATION_TASK = 'ai.terok.task' module-attribute

ANNOTATION_TASK_META_PATH = 'ai.terok.task_meta_path' module-attribute

IdentityResolver(inspector)

Compose a ContainerInspector lookup + task-meta YAML into an identity.

Callable: resolver(container_id) -> ContainerIdentity. Four soft-fail paths, all returning a degraded identity that keeps the notification pipeline usable:

  • The inspector failed → empty ContainerIdentity; the subscriber falls back to the raw container ID.
  • Container carries no terok annotations (a standalone container that happened to hit the firewall) → container-name-only.
  • ai.terok.task_meta_path annotation absent → identity without task_name (project + task_id still present).
  • task_meta_path YAML unreadable / missing / malformed → same as above; the name field is left empty.

Configure the resolver with a ContainerInspector implementation.

The inspector is required (no default) so the caller owns the runtime-selection decision — clearance is runtime-neutral and must not reach for a specific backend itself. The notifier entry point picks an appropriate implementation at startup (terok-sandbox's create_container_inspector when available, NullInspector otherwise).

Source code in src/terok_clearance/client/identity_resolver.py 
73
74
75
76
77
78
79
80
81
82
83
def __init__(self, inspector: ContainerInspector) -> None:
    """Configure the resolver with a [`ContainerInspector`][terok_clearance.client.identity_resolver.ContainerInspector] implementation.

    The inspector is required (no default) so the caller owns the
    runtime-selection decision — clearance is runtime-neutral and
    must not reach for a specific backend itself.  The notifier
    entry point picks an appropriate implementation at startup
    (terok-sandbox's ``create_container_inspector`` when available,
    [`NullInspector`][terok_clearance.NullInspector] otherwise).
    """
    self._inspector = inspector

__call__(container_id)

Return the task-aware identity for container_id.

Source code in src/terok_clearance/client/identity_resolver.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def __call__(self, container_id: str) -> ContainerIdentity:
    """Return the task-aware identity for *container_id*."""
    try:
        info = self._inspector(container_id)
    except Exception:
        # The contract says inspectors soft-fail by returning an
        # empty ``ContainerInfo``, but a runtime-side race or an
        # unexpected error path in a third-party backend can still
        # raise.  Clamp it here so the caller (notifier / TUI)
        # never takes a crash from identity resolution.
        _log.debug("ContainerInspector raised for %s", container_id, exc_info=True)
        return ContainerIdentity()
    if not info.container_id:
        return ContainerIdentity()
    project = info.annotations.get(ANNOTATION_PROJECT, "")
    task_id = info.annotations.get(ANNOTATION_TASK, "")
    if not (project and task_id):
        return ContainerIdentity(container_name=info.name, project=project, task_id=task_id)
    meta_path = info.annotations.get(ANNOTATION_TASK_META_PATH, "")
    return ContainerIdentity(
        container_name=info.name,
        project=project,
        task_id=task_id,
        task_name=_read_task_name(meta_path) if meta_path else "",
    )