stuff

src/wf_artifacts/__init__.py

@@ -1,3 +1,8 @@
+from .catalog import (
+    WorkflowArtifactCatalogEntry,
+    artifact_catalog_entry,
+    artifact_node_name,
+)
 from .models import (
     AvailableCapability,
     AvailableSource,
@@ -20,7 +25,10 @@ __all__ = [
     "FileWorkflowArtifactStore",
     "RequiredCapability",
     "WorkflowArtifact",
+    "WorkflowArtifactCatalogEntry",
     "WorkflowArtifactStore",
     "WorkflowDeployment",
+    "artifact_catalog_entry",
+    "artifact_node_name",
     "validate_deployment_dependencies",
 ]

src/wf_artifacts/catalog.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from .models import DependencyDiagnostic, JsonObject, WorkflowArtifact
+
+
+class WorkflowArtifactCatalogEntry(BaseModel):
+    """NodeSpec-shaped projection of a saved workflow artifact."""
+
+    name: str
+    display_name: str
+    description: str | None = None
+    outcomes: tuple[str, ...]
+    input_schema: JsonObject
+    output_schema: JsonObject
+    required_sources: tuple[str, ...] = Field(default_factory=tuple)
+    diagnostics: tuple[DependencyDiagnostic, ...] = Field(default_factory=tuple)
+
+
+def artifact_node_name(artifact: WorkflowArtifact) -> str:
+    """Return the stable planner name for an artifact version."""
+    return f"workflow.{artifact.id}.v{artifact.version}"
+
+
+def artifact_catalog_entry(
+    artifact: WorkflowArtifact,
+    *,
+    diagnostics: list[DependencyDiagnostic] | tuple[DependencyDiagnostic, ...] = (),
+) -> WorkflowArtifactCatalogEntry:
+    """Project an artifact as a catalog entry without exposing its internal plan."""
+    required_sources = sorted(
+        {
+            capability.logical_source
+            for capability in artifact.required_capabilities.values()
+        }
+    )
+    return WorkflowArtifactCatalogEntry(
+        name=artifact_node_name(artifact),
+        display_name=artifact.title,
+        description=artifact.description,
+        outcomes=artifact.outcomes,
+        input_schema=artifact.input_schema,
+        output_schema=artifact.output_schema,
+        required_sources=tuple(required_sources),
+        diagnostics=tuple(diagnostics),
+    )

tests/artifacts/test_catalog.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from wf_artifacts import (
+    DependencyDiagnostic,
+    DiagnosticSeverity,
+    RequiredCapability,
+    WorkflowArtifact,
+    artifact_catalog_entry,
+    artifact_node_name,
+)
+
+
+def artifact() -> WorkflowArtifact:
+    return WorkflowArtifact(
+        id="summarize_docs",
+        version=2,
+        title="Summarize Docs",
+        description="Summarize retrieved documentation.",
+        input_schema={"type": "object", "properties": {"query": {"type": "string"}}},
+        output_schema={
+            "type": "object",
+            "properties": {"summary": {"type": "string"}},
+        },
+        outcomes=("done", "failed"),
+        plan={"name": "summarize_docs", "nodes": [{"id": "hidden"}]},
+        required_capabilities={
+            "context7.query-docs": RequiredCapability(
+                logical_source="context7",
+                capability_name="query-docs",
+                kind="tool",
+            )
+        },
+    )
+
+
+def test_artifact_node_name_is_stable_and_versioned() -> None:
+    name = artifact_node_name(artifact())
+
+    assert name == "workflow.summarize_docs.v2"
+
+
+def test_artifact_catalog_entry_is_nodespec_shaped_without_internal_plan() -> None:
+    entry = artifact_catalog_entry(artifact())
+    dumped = entry.model_dump(mode="json")
+
+    assert dumped["name"] == "workflow.summarize_docs.v2"
+    assert dumped["display_name"] == "Summarize Docs"
+    assert dumped["description"] == "Summarize retrieved documentation."
+    assert dumped["outcomes"] == ["done", "failed"]
+    assert dumped["input_schema"]["properties"]["query"]["type"] == "string"
+    assert dumped["output_schema"]["properties"]["summary"]["type"] == "string"
+    assert dumped["required_sources"] == ["context7"]
+    assert "plan" not in dumped
+
+
+def test_artifact_catalog_entry_includes_dependency_diagnostics() -> None:
+    diagnostic = DependencyDiagnostic(
+        severity=DiagnosticSeverity.ERROR,
+        code="source_missing",
+        logical_ref="context7.query-docs",
+        bound_source="context7.personal",
+        message="Bound source is unavailable.",
+    )
+
+    entry = artifact_catalog_entry(artifact(), diagnostics=[diagnostic])
+    dumped = entry.model_dump(mode="json")
+
+    assert len(dumped["diagnostics"]) == 1
+    assert dumped["diagnostics"][0]["code"] == "source_missing"
+    assert dumped["diagnostics"][0]["logical_ref"] == "context7.query-docs"