orchestrata_ultimate.context.providers.artifact_digest

Reference › orchestrata_ultimate › context › providers › artifact_digest

This provider renders a compact, chronological digest of what upstream pipeline stages produced, so a downstream volk sees prior artifacts without copying payload keys by hand. The assembler uses it as the fallback when the artifact store has no digest method of its own. Tests are in tests/context/providers/test_artifact_digest.py. render_artifact_digest carries @traces for REQ-001-US-006-018.

It is store-agnostic by duck typing. The two artifact store implementations, in the kernel and the orchestrator, are normalised by _to_artifact_view into a small frozen _ArtifactView with stage, type, a 200-character content preview and a timestamp. _artifacts_from_store tries by_thread for the thread and falls back to all. The render sorts by timestamp when present, keeps the last max_items (default 10), and returns a markdown block headed with the prior-artifacts title, one line per artifact. A missing store or no artifacts gives an empty string.

At a glance

Class diagram

Dependency diagram

API

Classes

_ArtifactView (dataclass)

Decorators: dataclass(frozen=True) · Line 23

Normalized view of an artifact from either store implementation.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/context/providers/artifact_digest.py

"""Render a compact digest of prior-stage artifacts for volk prompts.

This provider closes Gap E from ``docs/context-injection.md``: downstream
volks receive a chronological summary of what upstream stages produced,
without relying on each volk manually copying the right payload keys.

It is intentionally store-agnostic: both ``kernel.pipeline_artifacts.ArtifactStore``
and ``orchestrator.artifact_store.ArtifactStore`` are supported by duck-typing
(``by_thread``, ``all``, ``stage``/``stage_alias``, ``type``/``artifact_type``).
Full canonicalization of the two stores is tracked as a later cleanup.
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from typing import Any

from orchestrata_ultimate.core.base.traces import traces


@dataclass(frozen=True)
class _ArtifactView:
    """Normalized view of an artifact from either store implementation."""

    stage: str
    type: str
    content_preview: str
    timestamp: datetime | None


def _getattr_str(obj: Any, *names: str) -> str:
    """Return the first present string attribute, or ''."""
    for name in names:
        value = getattr(obj, name, None)
        if value is not None:
            return str(value)
    return ""


def _to_artifact_view(artifact: Any) -> _ArtifactView:
    """Normalize either PipelineArtifact implementation to a view."""
    stage = _getattr_str(artifact, "stage", "stage_alias")
    type_ = _getattr_str(artifact, "type", "artifact_type")
    content = _getattr_str(artifact, "content")
    timestamp = getattr(artifact, "timestamp", None)
    return _ArtifactView(
        stage=stage,
        type=type_,
        content_preview=content[:200],
        timestamp=timestamp if isinstance(timestamp, datetime) else None,
    )


def _artifacts_from_store(store: Any, thread_id: str) -> list[Any]:
    """Return artifacts for *thread_id* if available, else all artifacts."""
    if thread_id and hasattr(store, "by_thread"):
        try:
            return list(store.by_thread(thread_id))
        except Exception:
            pass
    if hasattr(store, "all"):
        return list(store.all())
    return []


@traces(spec="001", reqs=["REQ-001-US-006-018"])
def render_artifact_digest(
    store: Any,
    thread_id: str = "",
    max_items: int = 10,
) -> str:
    """Render a compact chronological digest of artifacts.

    Args:
        store: Any artifact store with ``by_thread`` / ``all`` query methods.
        thread_id: Pipeline thread to filter on.
        max_items: Maximum artifacts to include.

    Returns:
        Markdown digest block or "" when no artifacts are available.
    """
    if store is None:
        return ""

    raw_artifacts = _artifacts_from_store(store, thread_id)
    if not raw_artifacts:
        return ""

    views = [_to_artifact_view(a) for a in raw_artifacts]
    # Sort by timestamp when present, otherwise preserve store order.
    views.sort(key=lambda v: (v.timestamp is None, v.timestamp or datetime.min))

    lines: list[str] = ["# PRIOR ARTIFACTS"]
    for view in views[-max_items:]:
        lines.append(f"- **{view.stage}** / `{view.type}`: {view.content_preview}")

    return "\n".join(lines)