Service-API completion: ingestion, retrieval, transformations, workflows, actor/delegation context, bounded agent operations, context packages, and dry-run/review-gate contracts

tests/test_service_api.py

@@ -1,7 +1,16 @@
 import pytest
 
-from kontextual_engine import AuthorizationError, OperationContext, PolicyDecision, ServiceRuntime, create_app
+from kontextual_engine import (
+    AuthorizationError,
+    OperationContext,
+    PolicyDecision,
+    PolicyEffect,
+    ServiceRuntime,
+    ValidationError,
+    create_app,
+)
 from kontextual_engine.adapters.memory import InMemoryAssetRegistryRepository
+from kontextual_engine.api.app import _authorization_error_payload
 
 
 def test_service_runtime_health_readiness_and_version_are_importable_without_fastapi() -> None:
@@ -115,6 +124,399 @@ def test_service_runtime_policy_denial_blocks_protected_asset_operation() -> Non
     assert runtime.list_audit_events(target="asset:asset-denied")["items"][0]["outcome"] == "denied"
 
 
+def test_service_runtime_exposes_ingestion_retrieval_transformation_and_workflow_operations(
+    tmp_path,
+) -> None:
+    runtime = ServiceRuntime(repository=InMemoryAssetRegistryRepository())
+    context = runtime.operation_context(actor_id="user-api", correlation_id="corr-jobs")
+    source = tmp_path / "kontextual-source.txt"
+    source.write_text(
+        "Kontextual engine captures markdown proxy context, system boundaries, and source-grounded retrieval.",
+        encoding="utf-8",
+    )
+
+    ingestion = runtime.start_ingestion_job(
+        {
+            "mode": "file",
+            "path": str(source),
+            "asset_id": "asset-ingested",
+            "classification": {
+                "asset_type": "document",
+                "sensitivity": "internal",
+                "owner": "Platform Knowledge",
+                "topics": ["retrieval", "workflow"],
+            },
+        },
+        context,
+    )
+    retrieval = runtime.query_assets(
+        {
+            "text": "proxy context",
+            "include_snippets": True,
+            "representation_kind": "normalized",
+            "limit": 5,
+        },
+        context,
+    )
+    transformation = runtime.execute_transformation(
+        {
+            "operation_id": "structured_view",
+            "source_asset_ids": ["asset-ingested"],
+            "output_asset_id": "asset-derived",
+            "output_title": "Structured API View",
+            "metadata": {"request": "service-api-test"},
+        },
+        context,
+    )
+    template = runtime.register_workflow_template(
+        {
+            "template_id": "workflow-structured",
+            "name": "Structured Workflow",
+            "inputs": [{"name": "source_asset_ids", "kind": "asset"}],
+            "steps": [
+                {
+                    "step_id": "build",
+                    "kind": "transformation",
+                    "operation_id": "structured_view",
+                    "inputs": {"source_asset_ids": "$inputs.source_asset_ids"},
+                    "outputs": {
+                        "asset_id": "asset-workflow-structured",
+                        "title": "Workflow Structured View",
+                        "asset_type": "derived_artifact",
+                        "media_type": "application/json",
+                    },
+                }
+            ],
+        },
+        context,
+    )
+    workflow = runtime.invoke_workflow_run(
+        {
+            "template_id": template["template_id"],
+            "inputs": {"source_asset_ids": ["asset-ingested"]},
+        },
+        context,
+    )
+    reconstruction = runtime.reconstruct_workflow_run(workflow["run"]["run_id"])
+
+    assert ingestion["status"] == "completed"
+    assert ingestion["correlation_id"] == "corr-jobs"
+    assert ingestion["output_asset_ids"] == ["asset-ingested"]
+    assert ingestion["retry_options"]["retryable"] is False
+    assert ingestion["job"]["partial_results"]["action"] == "created"
+    assert runtime.ingestion_capabilities()["connectors"][0]["supports_directories"] is True
+
+    assert retrieval["success"] is True
+    assert retrieval["correlation_id"] == "corr-jobs"
+    assert retrieval["results"][0]["asset_id"] == "asset-ingested"
+    assert retrieval["results"][0]["source_refs"]
+    assert retrieval["results"][0]["snippets"][0]["source_ref_id"]
+    assert retrieval["metadata"]["policy_enforced"] is True
+
+    assert transformation["success"] is True
+    assert transformation["run"]["status"] == "completed"
+    assert transformation["run"]["output_asset_ids"] == ["asset-derived"]
+    assert transformation["lineage"]["source_asset_ids"] == ["asset-ingested"]
+    assert transformation["audit_event"]["operation"] == "transformation.run.completed"
+    assert transformation["retry_options"]["retryable"] is False
+    assert runtime.get_transformation_run(transformation["run"]["run_id"])["status"] == "completed"
+    assert runtime.list_transformation_operations()["count"] >= 1
+    assert runtime.list_transformation_runs(status="completed")["count"] >= 2
+
+    assert workflow["success"] is True
+    assert workflow["run"]["status"] == "completed"
+    assert workflow["run"]["output_asset_ids"] == ["asset-workflow-structured"]
+    assert workflow["retry_options"]["retryable"] is False
+    assert runtime.get_workflow_run(workflow["run"]["run_id"])["status"] == "completed"
+    assert runtime.list_workflow_runs(status="completed")["count"] == 1
+    assert runtime.list_workflow_review_tasks(workflow_run_id=workflow["run"]["run_id"])["count"] == 0
+    assert runtime.list_workflow_exceptions(workflow_run_id=workflow["run"]["run_id"])["count"] == 0
+    assert reconstruction["run"]["run_id"] == workflow["run"]["run_id"]
+    assert reconstruction["template"]["template_id"] == "workflow-structured"
+    assert reconstruction["transformation_runs"][0]["status"] == "completed"
+    assert reconstruction["derived_lineage"][0]["output_asset_id"] == "asset-workflow-structured"
+
+
+def test_service_runtime_exposes_failed_ingestion_job_state_and_recovery_envelope(tmp_path) -> None:
+    runtime = ServiceRuntime(repository=InMemoryAssetRegistryRepository())
+    context = runtime.operation_context(actor_id="user-api", correlation_id="corr-failed-job")
+
+    failed = runtime.start_ingestion_job(
+        {"mode": "file", "path": str(tmp_path / "missing.txt")},
+        context,
+    )
+    listed = runtime.list_ingestion_jobs(status="failed")
+
+    assert failed["status"] == "failed"
+    assert failed["correlation_id"] == "corr-failed-job"
+    assert failed["failures"][0]["code"] == "kontextual.not_found"
+    assert failed["retry_options"]["retryable"] is False
+    assert runtime.get_ingestion_job(failed["job_id"])["job_id"] == failed["job_id"]
+    assert listed["count"] == 1
+    assert listed["items"][0]["job_id"] == failed["job_id"]
+
+
+def test_service_runtime_operation_context_represents_delegation_and_agent_identity() -> None:
+    runtime = ServiceRuntime(repository=InMemoryAssetRegistryRepository())
+
+    context = runtime.operation_context(
+        actor_id="agent-codex",
+        actor_type="ai_agent",
+        display_name="Codex",
+        external_ref="agent://codex",
+        groups=["automation", "engineering"],
+        delegated_actor_id="user-owner",
+        delegated_actor_type="human",
+        delegated_actor_display_name="Owner",
+        delegated_actor_groups=["knowledge-owners"],
+        request_scope={"tenant": "tenant-a", "surface": "service-api"},
+        policy_scope={"allowed_sensitivity": "internal"},
+        agent_id="agent-codex",
+        agent_name="Codex",
+        agent_run_id="run-123",
+        agent_tool="service-api",
+        correlation_id="corr-delegated",
+    )
+    payload = context.to_dict()
+
+    assert payload["actor"]["id"] == "agent-codex"
+    assert payload["actor"]["actor_type"] == "ai_agent"
+    assert payload["actor"]["groups"] == ["automation", "engineering"]
+    assert payload["delegated_actor"]["id"] == "user-owner"
+    assert payload["delegated_actor"]["actor_type"] == "human"
+    assert payload["request_scope"]["tenant"] == "tenant-a"
+    assert payload["policy_scope"]["allowed_sensitivity"] == "internal"
+    assert payload["metadata"]["agent"]["agent_run_id"] == "run-123"
+    assert payload["metadata"]["delegation"]["delegated_actor_id"] == "user-owner"
+
+
+def test_service_api_authorization_error_payload_redacts_resource_metadata() -> None:
+    runtime = ServiceRuntime(
+        repository=InMemoryAssetRegistryRepository(),
+        policy_gateway=DenyCreatePolicy(),
+    )
+
+    with pytest.raises(AuthorizationError) as exc_info:
+        runtime.create_asset(
+            {
+                "asset_id": "asset-secret",
+                "title": "Protected Title",
+                "classification": {
+                    "asset_type": "document",
+                    "sensitivity": "restricted",
+                    "metadata": {"secret": "do-not-echo"},
+                },
+            },
+            runtime.operation_context(actor_id="user-denied"),
+        )
+
+    payload = _authorization_error_payload(exc_info.value)
+
+    assert payload["details"]["policy_decision"]["effect"] == "deny"
+    assert "resource_metadata" not in payload["details"]["policy_decision"].get("context", {})
+    assert "Protected Title" not in str(payload)
+    assert "do-not-echo" not in str(payload)
+
+
+def test_service_runtime_exposes_bounded_agent_operation_catalog_and_dispatch() -> None:
+    runtime = ServiceRuntime(repository=InMemoryAssetRegistryRepository())
+    context = runtime.operation_context(
+        actor_id="agent-codex",
+        actor_type="ai_agent",
+        agent_id="agent-codex",
+        agent_run_id="run-agent-1",
+        correlation_id="corr-agent",
+    )
+    runtime.create_asset(
+        {
+            "asset_id": "asset-agent",
+            "title": "Agent Visible Asset",
+            "classification": {"asset_type": "document", "sensitivity": "internal"},
+            "source_refs": [{"source_system": "test", "path": "agent.md"}],
+        },
+        context,
+    )
+
+    catalog = runtime.list_agent_operations()
+    dry_run = runtime.execute_agent_operation(
+        "enrich_metadata",
+        {
+            "dry_run": True,
+            "payload": {"asset_id": "asset-agent", "metadata": {"key": "agent_note", "value": "planned"}},
+        },
+        context,
+    )
+    search = runtime.execute_agent_operation(
+        "search_assets",
+        {"query": {"asset_type": "document", "limit": 5}},
+        context,
+    )
+    bundle = runtime.execute_agent_operation(
+        "retrieve_asset",
+        {"asset_id": "asset-agent"},
+        context,
+    )
+    report = runtime.execute_agent_operation(
+        "report_result",
+        {"summary": "Agent operation completed", "result_ref": {"asset_id": "asset-agent"}},
+        context,
+    )
+
+    assert catalog["count"] >= 10
+    assert runtime.get_agent_operation("search_assets")["audit_operation"] == "agent.operation.search_assets"
+    assert all(item["required_permissions"] for item in catalog["items"])
+    assert all(item["failure_modes"] for item in catalog["items"])
+    assert dry_run["dry_run"] is True
+    assert runtime.list_metadata_records("asset-agent")["count"] == 0
+    assert search["success"] is True
+    assert search["result"]["results"][0]["asset_id"] == "asset-agent"
+    assert search["audit_event"]["operation"] == "agent.operation.search_assets"
+    assert bundle["result"]["asset"]["id"] == "asset-agent"
+    assert bundle["result"]["source_grounded"] is True
+    assert report["result"]["audit_event"]["operation"] == "agent.report.recorded"
+    assert runtime.list_audit_events(target="agent_operation:search_assets")["count"] == 2
+
+    with pytest.raises(ValidationError):
+        runtime.execute_agent_operation("shell", {"command": "echo no"}, context)
+
+
+def test_service_runtime_assembles_source_grounded_context_package_with_opaque_memory_refs(
+    tmp_path,
+) -> None:
+    runtime = ServiceRuntime(repository=InMemoryAssetRegistryRepository())
+    context = runtime.operation_context(actor_id="agent-codex", actor_type="ai_agent", correlation_id="corr-context")
+    source = tmp_path / "context-source.txt"
+    source.write_text(
+        "Markdown-backed context packages should remain source grounded.",
+        encoding="utf-8",
+    )
+    runtime.start_ingestion_job(
+        {
+            "mode": "file",
+            "path": str(source),
+            "asset_id": "asset-context",
+            "classification": {"asset_type": "markdown", "sensitivity": "internal"},
+        },
+        context,
+    )
+
+    package = runtime.assemble_context_package(
+        {
+            "package_id": "ctxpkg-test",
+            "title": "Context Package",
+            "intent": "Support implementation planning.",
+            "format": "markitect",
+            "query": {
+                "text": "source grounded",
+                "representation_kind": "normalized",
+                "include_snippets": True,
+            },
+            "constraints": {"max_sensitivity": "internal", "no_external_publish": True},
+            "external_memory_refs": [
+                {
+                    "ref_id": "phase-memory:episode-1",
+                    "system": "phase-memory",
+                    "content": "memory graph detail must not be embedded",
+                }
+            ],
+        },
+        context,
+    )
+
+    assert runtime.context_package_schema()["formats"] == ["kontextual", "markitect"]
+    assert package["package_id"] == "ctxpkg-test"
+    assert package["source_grounded"] is True
+    assert package["result_count"] == 1
+    assert package["items"][0]["asset_id"] == "asset-context"
+    assert package["items"][0]["source_refs"]
+    assert package["items"][0]["snippets"]
+    assert package["external_memory_refs"] == [
+        {
+            "ref_id": "phase-memory:episode-1",
+            "system": "phase-memory",
+            "kind": "memory_ref",
+            "opaque": True,
+            "metadata": {},
+        }
+    ]
+    assert "memory graph detail" not in str(package["external_memory_refs"])
+    assert package["markitect_payload"]["kind"] == "markitect.context_package"
+    assert package["markitect_payload"]["adapter_boundary"] == (
+        "markdown rendering and selector semantics are delegated to markitect-tool"
+    )
+    assert package["audit_event"]["operation"] == "context_package.assemble"
+    assert package["policy_decision"]["effect"] == "allow"
+
+
+def test_service_runtime_agent_operations_support_review_required_and_dry_run_only_policy() -> None:
+    review_runtime = ServiceRuntime(
+        repository=InMemoryAssetRegistryRepository(),
+        policy_gateway=ReviewRequiredAgentPolicy(),
+    )
+    dry_run_runtime = ServiceRuntime(
+        repository=InMemoryAssetRegistryRepository(),
+        policy_gateway=DryRunOnlyAgentPolicy(),
+    )
+    review_context = review_runtime.operation_context(
+        actor_id="agent-codex",
+        actor_type="ai_agent",
+        correlation_id="corr-review",
+    )
+    dry_run_context = dry_run_runtime.operation_context(
+        actor_id="agent-codex",
+        actor_type="ai_agent",
+        correlation_id="corr-dry-run",
+    )
+
+    review = review_runtime.execute_agent_operation(
+        "enrich_metadata",
+        {"payload": {"asset_id": "asset-review", "metadata": {"key": "risk", "value": "high"}}},
+        review_context,
+    )
+    dry_run_required = dry_run_runtime.execute_agent_operation(
+        "transform_asset",
+        {"payload": {"transformation": {"operation_id": "structured_view"}}},
+        dry_run_context,
+    )
+    dry_run = dry_run_runtime.execute_agent_operation(
+        "transform_asset",
+        {"dry_run": True, "payload": {"transformation": {"operation_id": "structured_view"}}},
+        dry_run_context,
+    )
+
+    assert review["success"] is False
+    assert review["review_required"] is True
+    assert review["audit_event"]["outcome"] == "review_required"
+    assert review["policy_decision"]["effect"] == "require_review"
+    assert dry_run_required["success"] is False
+    assert dry_run_required["dry_run_required"] is True
+    assert dry_run_required["audit_event"]["outcome"] == "dry_run"
+    assert dry_run_required["policy_decision"]["effect"] == "dry_run_only"
+    assert dry_run["success"] is True
+    assert dry_run["dry_run"] is True
+    assert dry_run["audit_event"]["outcome"] == "dry_run"
+
+
+def test_service_runtime_ingestion_directory_reports_partial_failures(tmp_path) -> None:
+    runtime = ServiceRuntime(repository=InMemoryAssetRegistryRepository())
+    context = runtime.operation_context(actor_id="user-api", correlation_id="corr-partial")
+    (tmp_path / "good.txt").write_text("source grounded text", encoding="utf-8")
+    (tmp_path / "unsupported.bin").write_bytes(b"\x00\x01\x02")
+
+    result = runtime.start_ingestion_job(
+        {"mode": "directory", "path": str(tmp_path), "recursive": False},
+        context,
+    )
+
+    assert result["status"] == "partially_completed"
+    assert result["output_asset_ids"]
+    assert result["failures"][0]["code"] == "kontextual.adapter_unavailable"
+    assert result["job"]["partial_results"]["files_total"] == 2
+    assert result["job"]["partial_results"]["failed"] == 1
+    assert result["job"]["partial_results"]["succeeded"] == 1
+    assert result["retry_options"]["retryable"] is True
+
+
 def test_create_app_reports_missing_optional_dependency_when_fastapi_is_absent() -> None:
     try:
         import fastapi  # noqa: F401
@@ -160,10 +562,20 @@ def test_service_health_readiness_version_and_openapi_contracts(client) -> None:
     assert "/api/v1/health" in paths
     assert "/api/v1/ready" in paths
     assert "/api/v1/version" in paths
+    assert "/api/v1/context" in paths
     assert "/api/v1/assets" in paths
     assert "/api/v1/relationships" in paths
     assert "/api/v1/audit/events" in paths
     assert "/api/v1/policy/evaluate" in paths
+    assert "/api/v1/ingestion/jobs" in paths
+    assert "/api/v1/retrieval/assets" in paths
+    assert "/api/v1/transformations/runs" in paths
+    assert "/api/v1/workflows/templates" in paths
+    assert "/api/v1/workflows/runs" in paths
+    assert "/api/v1/workflows/reviews" in paths
+    assert "/api/v1/agents/operations" in paths
+    assert "/api/v1/context-packages" in paths
+    assert "/api/v1/context-packages/schema" in paths
 
 
 def test_create_app_attaches_runtime_to_application_state(client) -> None:
@@ -188,3 +600,46 @@ class DenyCreatePolicy:
                 context={"resource_metadata": resource_metadata or {}},
             )
         return PolicyDecision.allow(context.actor.id, action, resource)
+
+
+class ReviewRequiredAgentPolicy:
+    def authorize(
+        self,
+        context: OperationContext,
+        action: str,
+        resource: str,
+        *,
+        resource_metadata: dict[str, str] | None = None,
+    ) -> PolicyDecision:
+        if action == "agent.operation.enrich_metadata":
+            return PolicyDecision(
+                PolicyEffect.REQUIRE_REVIEW,
+                context.actor.id,
+                action,
+                resource,
+                reason="metadata enrichment requires review",
+                obligations={"queue": "knowledge-review"},
+                context={"resource_metadata": resource_metadata or {}},
+            )
+        return PolicyDecision.allow(context.actor.id, action, resource)
+
+
+class DryRunOnlyAgentPolicy:
+    def authorize(
+        self,
+        context: OperationContext,
+        action: str,
+        resource: str,
+        *,
+        resource_metadata: dict[str, str] | None = None,
+    ) -> PolicyDecision:
+        if action == "agent.operation.transform_asset":
+            return PolicyDecision(
+                PolicyEffect.DRY_RUN_ONLY,
+                context.actor.id,
+                action,
+                resource,
+                reason="transformation requires dry-run preview",
+                context={"resource_metadata": resource_metadata or {}},
+            )
+        return PolicyDecision.allow(context.actor.id, action, resource)