From d8a4b6b3c1c4e33d72bb213b27489440c4e2cce8 Mon Sep 17 00:00:00 2001 From: Jonathan Boniface Date: Tue, 1 Sep 2026 17:21:58 +0100 Subject: [PATCH] fix: files manually --- .../skills/code-review-and-quality/SKILL.md | 54 ++++++++++++ Dockerfile | 36 +++++++- README.md | 84 +++++++++++++++++-- app/agent.py | 39 +++++---- app/card.py | 4 +- app/governance.py | 2 +- app/kab_ingestion/__init__.py | 5 +- app/kab_ingestion/connectors/github.py | 34 +------- app/kab_ingestion/connectors/sharepoint.py | 24 +----- app/kab_ingestion/domain/models.py | 39 +-------- app/main.py | 56 ++++++++++++- app/nodes/__init__.py | 6 ++ app/nodes/deployment_node.py | 41 +++++++++ app/nodes/discovery_node.py | 25 ++++++ app/nodes/generation_node.py | 33 ++++++++ app/nodes/validation_node.py | 41 +++++++++ app/publication.py | 6 +- app/routes.py | 2 +- app/states/ingestion_state.py | 25 ++++++ app/triggers.py | 2 +- app/validation.py | 2 +- app/workflows/__init__.py | 3 + app/workflows/ingestion_workflow.py | 77 +++++++++++++++++ config/example.yaml | 22 ----- deploy/kubernetes.yaml | 15 ---- docs/contract.md | 3 + kab_ingestion/__init__.py | 33 +++++++- kab_ingestion/governance.py | 11 +++ kab_ingestion/models.py | 42 +++++++++- pyproject.toml | 2 +- {validation => scripts}/validate_delivery.py | 4 +- tests/conftest.py | 2 +- tests/test_ingestion.py | 2 +- tests/test_wiring.py | 2 +- tests/test_workflow.py | 56 +++++++++++++ validation/step-1-evidence.md | 38 --------- 36 files changed, 656 insertions(+), 216 deletions(-) create mode 100644 .agents/skills/code-review-and-quality/SKILL.md create mode 100644 app/nodes/deployment_node.py create mode 100644 app/nodes/discovery_node.py create mode 100644 app/nodes/generation_node.py create mode 100644 app/nodes/validation_node.py create mode 100644 app/states/ingestion_state.py create mode 100644 app/workflows/ingestion_workflow.py delete mode 100644 config/example.yaml delete mode 100644 deploy/kubernetes.yaml rename {validation => scripts}/validate_delivery.py (93%) create mode 100644 tests/test_workflow.py delete mode 100644 validation/step-1-evidence.md diff --git a/.agents/skills/code-review-and-quality/SKILL.md b/.agents/skills/code-review-and-quality/SKILL.md new file mode 100644 index 0000000..1a67856 --- /dev/null +++ b/.agents/skills/code-review-and-quality/SKILL.md @@ -0,0 +1,54 @@ +--- +name: code-review-and-quality +description: Reviews agent changes along Standards and Spec axes. Use when reviewing PRs, auditing generated agents, or checking platform harness compliance. +--- + +# Code Review and Quality Assurance + +## Overview + +Two-axis review process for evaluating platform agent changes. Evaluates code along two independent dimensions — **Standards** (compliance with platform harness standards & coding guidelines) and **Spec** (faithful implementation of issue/PRD contracts). + +## When to Use + +- Auditing newly generated or refactored `crucible-agent-*` repositories +- Reviewing pull requests, git diffs, or feature branches +- Checking compliance with platform harness contracts (`execution-contract.md`, `base-state-contract.md`) + +--- + +## Two-Axis Review Framework + +### Axis 1: Standards (Compliance & Code Smells) +Inspect code against documented platform standards: + +- **Harness Boundary**: Verify `crucible-agent-factory-harness` vs `crucible-agents-sdk` boundary rules. No runnable runtime code inside harness standard repos. +- **Base State Schema**: Verify state schemas extend `BaseAgentState` (`messages`, `errors`, `has_errors`, `workflow_id`, `current_step`, `next_step`). +- **Container Hardening**: Verify multi-stage Dockerfiles, non-root execution (`UID 10001`), and `/health` probes. +- **Code Smell Check**: Look for Fowler smells (Duplicated Code, Shotgun Surgery, Speculative Generality, Feature Envy). + +### Axis 2: Spec (Requirement & Contract Fidelity) +Inspect code against the originating specification (`spec/.md`): + +- **Missing/Partial Requirements**: Are all contract endpoints, connector types, and trigger modes fully implemented? +- **Scope Creep**: Was functionality added that was not requested in the contract spec? +- **Incorrect Behavior**: Do connector error handling, cursor advancement, or tenant isolation checks deviate from the spec contract? + +--- + +## Review Output Format + +```markdown +## Standards Review +- [Pass/Fail] Harness & Base State Compliance +- [Pass/Fail] Container & Security Hardening +- Findings & Code Smells + +## Spec Review +- [Pass/Fail] Contract & Requirement Coverage +- Findings & Scope Creep + +## Worst Issue Per Axis +- Standards: +- Spec: +``` diff --git a/Dockerfile b/Dockerfile index 1ad0afd..9b499fe 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,35 @@ -FROM python:3.11-slim +FROM python:3.11-slim AS builder + WORKDIR /app + +# Install dependencies and build wheel +COPY pyproject.toml requirements.txt ./ +RUN pip install --no-cache-dir --prefix=/install -r requirements.txt + +# Final production stage +FROM python:3.11-slim + +ENV PYTHONUNBUFFERED=1 \ + PYTHONDONTWRITEBYTECODE=1 \ + PORT=8080 + +WORKDIR /app + +# Create non-root user for security +RUN groupadd -g 10001 appgroup && \ + useradd -u 10001 -g appgroup -s /bin/sh -m appuser + +# Copy installed dependencies and application code +COPY --from=builder /install /usr/local COPY . . -RUN pip install --no-cache-dir . -CMD ["python", "-m", "kab_ingestion"] + +RUN chown -R appuser:appgroup /app + +USER appuser + +EXPOSE 8080 + +HEALTHCHECK --interval=30s --timeout=5s --start-period=5s --retries=3 \ + CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')" || exit 1 + +CMD ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"] diff --git a/README.md b/README.md index f744a96..a2351a7 100644 --- a/README.md +++ b/README.md @@ -1,14 +1,82 @@ # KAB Content Ingestion Agent -Contract-first ingestion for GitHub and SharePoint into a governed KAB-compatible knowledge store. The package is dependency-light and uses injected HTTP, secret, clock, and publisher ports for deterministic tests. +Contract-first, multi-tenant ingestion for GitHub REST and Microsoft Graph SharePoint into a governed KAB-compatible knowledge store. -## Quick start -```bash -python -m pytest -q -python -m compileall kab_ingestion +The package is dependency-light, utilizing hexagonal architecture ports (HTTP, SecretStore, Publisher) for deterministic testing and zero-token leakage. + +--- + +## Architecture & Codebase Alignment + +The codebase is organized into two primary layers with a unified domain alignment: + +``` + +-----------------------------------+ + | app/ | + | (Starlette/FastAPI Web Layer & | + | LangGraph Execution Workflows) | + +-----------------+-----------------+ + | + Imports & Delegates + v + +-----------------------------------+ + | kab_ingestion/ | + | (Core Domain Models, Connectors, | + | Orchestrator & Governance) | + +-----------------------------------+ ``` -See `docs/contract.md`, `docs/operations.md`, and `examples/config.yaml`. +### 1. Core Ingestion Domain Layer (`kab_ingestion/`) +- **[kab_ingestion/models.py](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/kab_ingestion/models.py)**: Canonical domain models (`Document`, `ACL`, `Provenance`, `SyncCursor`, `IngestionRequest`, `IngestionResult`). +- **[kab_ingestion/github.py](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/kab_ingestion/github.py)** & **[kab_ingestion/sharepoint.py](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/kab_ingestion/sharepoint.py)**: Source connectors supporting full tree syncs, incremental delta syncs, and webhook event parsing. +- **[kab_ingestion/orchestrator.py](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/kab_ingestion/orchestrator.py)**: Fault-tolerant batch sync engine with bounded retries (`max_retries=3`) and dead-lettering. +- **[kab_ingestion/governance.py](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/kab_ingestion/governance.py)**: Strict multi-tenant isolation and ACL validation. -## Security -Tokens are referenced by secret name, never stored in configuration or logs. Every document carries tenant, source, and ACL metadata; publishers must enforce the same filters at query time. +### 2. Application & Workflow Layer (`app/`) +- **[app/agent.py](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/app/agent.py)**: Service factory `create_agent_app()` and `IngestionService`. +- **[app/card.py](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/app/card.py)**: Agent discovery card (`AGENT_CARD`). +- **[app/workflows/ingestion_workflow.py](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/app/workflows/ingestion_workflow.py)**: 4-Phase LangGraph workflow runner. +- **[app/routes.py](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/app/routes.py)**: HTTP REST routes (`/health`, `/card`, `/ingest`). + +### 3. Package Re-export Alignment (`app/kab_ingestion`) +- **`app/kab_ingestion`**: Operates as a lightweight re-export alignment layer. All domain models, connectors, and ports re-export directly from the canonical `kab_ingestion` root package, guaranteeing full backward compatibility and preventing duplicate definitions. + +--- + +## LangGraph 4-Phase Execution Contract + +Every ingestion workflow implements the platform's 4-phase execution contract: + +1. **Discovery** (`app/nodes/discovery_node.py`): Non-blocking check for existing resources and connector availability. +2. **Validation** (`app/nodes/validation_node.py`): Validates `IngestionRequest` format and allowed sources; fails fast on error (`has_errors: True` → `END`). +3. **Generation** (`app/nodes/generation_node.py`): Connects to GitHub/SharePoint source connectors and fetches normalized documents into `fetched_documents`. +4. **Deployment** (`app/nodes/deployment_node.py`): Enforces governance approval and tenant ACL isolation, publishing approved documents to target storage. + +--- + +## Security & Tenant Isolation + +- **Secret References (`secret_ref`)**: Tokens are referenced by secret name (e.g. `secrets/github-read`), never stored in configuration files or logs. +- **Tenant Isolation & ACLs**: Every document carries `tenant_id` and `ACL {principals, groups, visibility}`. Tenant mismatch results in automatic publish rejection. + +--- + +## Quick Start & Testing + +### Running Tests +```bash +python -m pytest -q +``` + +### Compiling Packages +```bash +python -m compileall kab_ingestion app +``` + +--- + +## Documentation References + +- **Contract Specification**: [spec/ingestion-contract.md](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/spec/ingestion-contract.md) +- **Configuration Example**: [examples/config.yaml](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/examples/config.yaml) +- **Kubernetes Manifests**: [k8s/](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/k8s) diff --git a/app/agent.py b/app/agent.py index 703a859..32ccf26 100644 --- a/app/agent.py +++ b/app/agent.py @@ -3,10 +3,10 @@ from __future__ import annotations from dataclasses import dataclass from typing import Any +from kab_ingestion.github import GitHubConnector +from kab_ingestion.models import IngestionRequest, IngestionResult +from kab_ingestion.sharepoint import SharePointConnector from .card import AGENT_CARD, AgentCard -from .kab_ingestion.connectors.github import GitHubSCMConnector -from .kab_ingestion.connectors.sharepoint import SharePointCMSConnector -from .kab_ingestion.domain.models import IngestionRequest, IngestionResult from .publication import publish_document from .settings import Settings, get_settings from .validation import validate_document @@ -19,22 +19,27 @@ class IngestionService: cms: Any = None def ingest(self, request: IngestionRequest) -> list[IngestionResult]: - if request.source not in self.settings.allowed_sources: - return [] - results = [] - for document in self.scm.fetch(request): - report = validate_document(document) - if not report.valid: - from .kab_ingestion.domain.models import IngestionResult - results.append(IngestionResult(document, False, "invalid", "; ".join(report.errors))) - continue - results.append(publish_document(document, self.cms, self.settings.publish_enabled, self.settings.require_governance_approval)) - return results + from .workflows import create_ingestion_workflow + from .states.ingestion_state import IngestionState + + workflow = create_ingestion_workflow( + scm=self.scm, + cms=self.cms, + allowed_sources=self.settings.allowed_sources, + publish_enabled=self.settings.publish_enabled, + require_governance_approval=self.settings.require_governance_approval, + ) + initial_state: IngestionState = { + "request": request, + "tenant_id": request.metadata.get("tenant_id", "default_tenant"), + } + final_state = workflow.invoke(initial_state) + return final_state.get("published_results", []) def create_agent_app(settings: Settings | None = None) -> dict[str, Any]: config = settings or get_settings() - scm = GitHubSCMConnector(config.github_token, config.github_api_base) - cms = SharePointCMSConnector(config.sharepoint_site_url, config.sharepoint_access_token) if config.sharepoint_site_url and config.sharepoint_access_token else None + scm = GitHubConnector(config.github_token, config.github_api_base) + cms = SharePointConnector(config.sharepoint_site_url, config.sharepoint_access_token) if config.sharepoint_site_url and config.sharepoint_access_token else None service = IngestionService(config, scm, cms) - return {"card": AGENT_CARD, "service": service, "routes": {"/health": "health", "/card": "card", "/ingest": "ingest"}, "triggers": {"github": "github_trigger"}} + return {"card": AGENT_CARD, "service": service, "routes": {"/health": "health", "/card": "card", "/ingest": "ingest"}, "triggers": {"github_push": "github_trigger", "manual_ingest": "manual_trigger"}} diff --git a/app/card.py b/app/card.py index 224f698..9c4fac7 100644 --- a/app/card.py +++ b/app/card.py @@ -13,12 +13,12 @@ class AgentCard: name: str description: str version: str - skills: tuple[AgentSkill, ...] + skills: tuple[str, ...] AGENT_CARD = AgentCard( name="content-ingestion-agent", description="Ingest governed knowledge content from source control and publish it to a CMS.", version="1.0.0", - skills=(AgentSkill("ingest", "Ingest content", "Fetch and validate source documents."), AgentSkill("publish", "Publish content", "Apply governance and publish approved documents.")), + skills=("ingest_github", "publish_sharepoint", "validate_content"), ) diff --git a/app/governance.py b/app/governance.py index 1291c96..8e56023 100644 --- a/app/governance.py +++ b/app/governance.py @@ -1,6 +1,6 @@ from dataclasses import dataclass -from .kab_ingestion.domain.models import SourceDocument +from kab_ingestion.models import SourceDocument @dataclass(frozen=True) diff --git a/app/kab_ingestion/__init__.py b/app/kab_ingestion/__init__.py index bc2c715..27f3fe4 100644 --- a/app/kab_ingestion/__init__.py +++ b/app/kab_ingestion/__init__.py @@ -1,5 +1,4 @@ -"""Ingestion domain migrated into the application package.""" - -from .domain.models import IngestionRequest, IngestionResult, SourceDocument +"""Re-export canonical kab_ingestion symbols.""" +from kab_ingestion import IngestionRequest, IngestionResult, SourceDocument __all__ = ["IngestionRequest", "IngestionResult", "SourceDocument"] diff --git a/app/kab_ingestion/connectors/github.py b/app/kab_ingestion/connectors/github.py index 66ff0e8..ab90af2 100644 --- a/app/kab_ingestion/connectors/github.py +++ b/app/kab_ingestion/connectors/github.py @@ -1,32 +1,4 @@ -from __future__ import annotations +"""Re-export canonical GitHubConnector from kab_ingestion.github.""" +from kab_ingestion.github import GitHubConnector as GitHubSCMConnector -import json -from dataclasses import dataclass -from urllib.request import Request, urlopen - -from ..domain.models import IngestionRequest, SourceDocument - - -@dataclass -class GitHubSCMConnector: - token: str | None = None - api_base: str = "https://api.github.com" - - def fetch(self, request: IngestionRequest) -> list[SourceDocument]: - """Fetch a file or repository contents using the GitHub contents API.""" - url = request.locator if request.locator.startswith("http") else f"{self.api_base}/{request.locator.lstrip('/')}" - headers = {"Accept": "application/vnd.github+json", "User-Agent": "content-ingestion-agent"} - if self.token: - headers["Authorization"] = f"Bearer {self.token}" - with urlopen(Request(url, headers=headers), timeout=20) as response: - payload = json.loads(response.read().decode("utf-8")) - items = payload if isinstance(payload, list) else [payload] - return [self._document(item, request) for item in items if item.get("type", "file") == "file"] - - @staticmethod - def _document(item: dict, request: IngestionRequest) -> SourceDocument: - body = item.get("content", "") - if item.get("encoding") == "base64": - import base64 - body = base64.b64decode(body).decode("utf-8", errors="replace") - return SourceDocument(identifier=str(item.get("sha", item.get("path", "unknown"))), title=item.get("name", item.get("path", "Untitled")), body=body, source=request.source, url=item.get("html_url"), metadata={"path": item.get("path")}) +__all__ = ["GitHubSCMConnector"] diff --git a/app/kab_ingestion/connectors/sharepoint.py b/app/kab_ingestion/connectors/sharepoint.py index c147add..8424627 100644 --- a/app/kab_ingestion/connectors/sharepoint.py +++ b/app/kab_ingestion/connectors/sharepoint.py @@ -1,22 +1,4 @@ -from __future__ import annotations +"""Re-export canonical SharePointConnector from kab_ingestion.sharepoint.""" +from kab_ingestion.sharepoint import SharePointConnector as SharePointCMSConnector -import json -from dataclasses import dataclass -from urllib.request import Request, urlopen - -from ..domain.models import SourceDocument - - -@dataclass -class SharePointCMSConnector: - site_url: str - access_token: str - - def publish(self, document: SourceDocument) -> str: - """Create a SharePoint list item through Microsoft Graph.""" - endpoint = f"{self.site_url.rstrip('/')}/_api/web/lists/getbytitle('Knowledge')/items" - payload = json.dumps({"Title": document.title, "Content": document.body}).encode() - request = Request(endpoint, data=payload, method="POST", headers={"Authorization": f"Bearer {self.access_token}", "Accept": "application/json", "Content-Type": "application/json"}) - with urlopen(request, timeout=20) as response: - result = json.loads(response.read().decode("utf-8")) - return result.get("d", {}).get("Id", result.get("id", document.identifier)) +__all__ = ["SharePointCMSConnector"] diff --git a/app/kab_ingestion/domain/models.py b/app/kab_ingestion/domain/models.py index 314d3d2..fc28c19 100644 --- a/app/kab_ingestion/domain/models.py +++ b/app/kab_ingestion/domain/models.py @@ -1,37 +1,4 @@ -from __future__ import annotations +"""Re-export canonical models from kab_ingestion.models for backward compatibility.""" +from kab_ingestion.models import IngestionRequest, IngestionResult, SourceDocument -from dataclasses import dataclass, field -from datetime import datetime, timezone -from typing import Any - - -def utc_now() -> datetime: - return datetime.now(timezone.utc) - - -@dataclass(frozen=True) -class IngestionRequest: - source: str - locator: str - requested_by: str = "system" - metadata: dict[str, Any] = field(default_factory=dict) - - -@dataclass(frozen=True) -class SourceDocument: - identifier: str - title: str - body: str - source: str - url: str | None = None - metadata: dict[str, Any] = field(default_factory=dict) - fetched_at: datetime = field(default_factory=utc_now) - - -@dataclass(frozen=True) -class IngestionResult: - document: SourceDocument - accepted: bool - status: str - reason: str | None = None - published_url: str | None = None +__all__ = ["IngestionRequest", "IngestionResult", "SourceDocument"] diff --git a/app/main.py b/app/main.py index 511c04e..55379da 100644 --- a/app/main.py +++ b/app/main.py @@ -1,11 +1,59 @@ -from .agent import create_agent_app +from __future__ import annotations -app = create_agent_app() +import os +from typing import Any +from .agent import create_agent_app +from .card import AGENT_CARD +from .routes import card_route, health_route, ingest_route + +agent_context = create_agent_app() + + +def make_asgi_app() -> Any: + try: + from fastapi import FastAPI, Request + from fastapi.responses import JSONResponse + + asgi_app = FastAPI(title="content-ingestion-agent", version="1.0.0") + + @asgi_app.get("/health") + def health() -> dict[str, str]: + return health_route() + + @asgi_app.get("/card") + def card() -> Any: + return card_route(AGENT_CARD) + + @asgi_app.post("/ingest") + async def ingest(request: Request) -> dict[str, Any]: + from kab_ingestion.models import IngestionRequest + + data = await request.json() + req = IngestionRequest( + source=data.get("source", "github"), + locator=data.get("locator", ""), + requested_by=data.get("requested_by", "system"), + metadata=data.get("metadata", {}), + ) + results = ingest_route(req, agent_context["service"]) + return {"results": results, "status": "completed"} + + return asgi_app + except ImportError: + return agent_context + + +app = make_asgi_app() def main() -> None: - """Application entry point for ASGI hosts and local smoke checks.""" - print(app["card"].name) + port = int(os.environ.get("PORT", "8080")) + try: + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=port) + except ImportError: + print(f"Starting {AGENT_CARD.name} v{AGENT_CARD.version}") if __name__ == "__main__": diff --git a/app/nodes/__init__.py b/app/nodes/__init__.py index e69de29..ce77ebf 100644 --- a/app/nodes/__init__.py +++ b/app/nodes/__init__.py @@ -0,0 +1,6 @@ +from .discovery_node import discovery_node +from .validation_node import validation_node +from .generation_node import generation_node +from .deployment_node import deployment_node + +__all__ = ["discovery_node", "validation_node", "generation_node", "deployment_node"] diff --git a/app/nodes/deployment_node.py b/app/nodes/deployment_node.py new file mode 100644 index 0000000..cc0b1ab --- /dev/null +++ b/app/nodes/deployment_node.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +from typing import Any +from kab_ingestion.models import IngestionResult +from ..publication import publish_document +from ..states.ingestion_state import IngestionState +from ..validation import validate_document + + +def deployment_node( + state: IngestionState, + cms: Any = None, + publish_enabled: bool = False, + require_governance_approval: bool = True, +) -> dict[str, Any]: + """Deployment Phase: Review governance, enforce tenant ACL isolation, and publish documents.""" + documents = state.get("fetched_documents", []) + results: list[IngestionResult] = [] + + for doc in documents: + report = validate_document(doc) + if not report.valid: + results.append(IngestionResult(doc, False, "invalid", "; ".join(report.errors))) + continue + res = publish_document(doc, cms, publish_enabled, require_governance_approval) + results.append(res) + + accepted_count = sum(1 for r in results if r.accepted) + final_output = { + "total": len(results), + "accepted": accepted_count, + "results": results, + "status": "completed" if not state.get("has_errors") else "completed_with_errors", + } + + return { + "published_results": results, + "final_output": final_output, + "current_step": "deployment", + "next_step": "END", + } diff --git a/app/nodes/discovery_node.py b/app/nodes/discovery_node.py new file mode 100644 index 0000000..5768686 --- /dev/null +++ b/app/nodes/discovery_node.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from typing import Any +from ..states.ingestion_state import IngestionState + + +def discovery_node(state: IngestionState) -> dict[str, Any]: + """Discovery Phase: Check existing capabilities and resources non-blockingly.""" + request = state.get("request") + source = request.source if request else "unknown" + locator = request.locator if request else "unknown" + + # Non-blocking check for existing ingestion resources + existing_resources = { + "source": source, + "locator": locator, + "previously_ingested": False, + "connector_available": source in {"github", "sharepoint"}, + } + + return { + "existing_resources": existing_resources, + "current_step": "discovery", + "next_step": "validation", + } diff --git a/app/nodes/generation_node.py b/app/nodes/generation_node.py new file mode 100644 index 0000000..6767de9 --- /dev/null +++ b/app/nodes/generation_node.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +from typing import Any +from ..states.ingestion_state import IngestionState + + +def generation_node(state: IngestionState, scm: Any = None) -> dict[str, Any]: + """Generation Phase: Fetch and normalize documents from source connector.""" + request = state.get("request") + if not request or not scm: + return { + "fetched_documents": [], + "current_step": "generation", + "next_step": "deployment", + } + + try: + documents = list(scm.fetch(request)) + return { + "fetched_documents": documents, + "current_step": "generation", + "next_step": "deployment", + } + except Exception as exc: + error_records = list(state.get("errors", [])) + error_records.append({"step": "generation", "message": str(exc), "recoverable": True}) + return { + "fetched_documents": [], + "errors": error_records, + "has_errors": True, + "current_step": "generation", + "next_step": "END", + } diff --git a/app/nodes/validation_node.py b/app/nodes/validation_node.py new file mode 100644 index 0000000..b6c6132 --- /dev/null +++ b/app/nodes/validation_node.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +from typing import Any +from ..states.ingestion_state import IngestionState + + +def validation_node(state: IngestionState, allowed_sources: set[str] | None = None) -> dict[str, Any]: + """Validation Phase: Validate request input and fail fast on malformed requests.""" + request = state.get("request") + allowed = allowed_sources or {"github", "sharepoint"} + errors: list[str] = [] + + if not request: + errors.append("missing ingestion request") + else: + if not request.source or not request.source.strip(): + errors.append("missing request source") + elif request.source not in allowed: + errors.append(f"unsupported source '{request.source}'") + + if not request.locator or not request.locator.strip(): + errors.append("missing request locator") + + if errors: + error_records = list(state.get("errors", [])) + for err in errors: + error_records.append({"step": "validation", "message": err, "recoverable": False}) + + return { + "validation_errors": errors, + "errors": error_records, + "has_errors": True, + "current_step": "validation", + "next_step": "END", + } + + return { + "validation_errors": [], + "current_step": "validation", + "next_step": "generation", + } diff --git a/app/publication.py b/app/publication.py index d0d0336..a075435 100644 --- a/app/publication.py +++ b/app/publication.py @@ -1,9 +1,9 @@ -from .kab_ingestion.domain.models import IngestionResult, SourceDocument -from .kab_ingestion.ports.cms import CMSConnector +from typing import Any +from kab_ingestion.models import IngestionResult, SourceDocument from .governance import review -def publish_document(document: SourceDocument, cms: CMSConnector | None, enabled: bool, require_approval: bool) -> IngestionResult: +def publish_document(document: SourceDocument, cms: Any | None, enabled: bool, require_approval: bool) -> IngestionResult: decision = review(document, require_approval) if not decision.approved: return IngestionResult(document, False, "rejected", decision.reason) diff --git a/app/routes.py b/app/routes.py index b103c91..0053d69 100644 --- a/app/routes.py +++ b/app/routes.py @@ -1,4 +1,4 @@ -from .kab_ingestion.domain.models import IngestionRequest +from kab_ingestion.models import IngestionRequest def ingest_route(request: IngestionRequest, service): diff --git a/app/states/ingestion_state.py b/app/states/ingestion_state.py new file mode 100644 index 0000000..96cfa67 --- /dev/null +++ b/app/states/ingestion_state.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from typing import Any, TypedDict +from kab_ingestion.models import Document, IngestionRequest, IngestionResult, SourceDocument + + +class IngestionState(TypedDict, total=False): + # Base agent state contract fields + messages: list[dict[str, Any]] + errors: list[dict[str, Any]] + has_errors: bool + workflow_id: str | None + current_step: str + next_step: str | None + started_at: str | None + updated_at: str | None + final_output: dict[str, Any] | None + + # Ingestion phase extension fields + request: IngestionRequest + tenant_id: str + existing_resources: dict[str, Any] + validation_errors: list[str] + fetched_documents: list[Document | SourceDocument] + published_results: list[IngestionResult] diff --git a/app/triggers.py b/app/triggers.py index 96ca407..88052ff 100644 --- a/app/triggers.py +++ b/app/triggers.py @@ -1,4 +1,4 @@ -from .kab_ingestion.domain.models import IngestionRequest +from kab_ingestion.models import IngestionRequest def github_trigger(locator: str, requested_by: str = "github-webhook") -> IngestionRequest: diff --git a/app/validation.py b/app/validation.py index b8cc09f..7e9fa1a 100644 --- a/app/validation.py +++ b/app/validation.py @@ -1,6 +1,6 @@ from dataclasses import dataclass -from .kab_ingestion.domain.models import SourceDocument +from kab_ingestion.models import SourceDocument @dataclass(frozen=True) diff --git a/app/workflows/__init__.py b/app/workflows/__init__.py index e69de29..c902f39 100644 --- a/app/workflows/__init__.py +++ b/app/workflows/__init__.py @@ -0,0 +1,3 @@ +from .ingestion_workflow import IngestionWorkflow, create_ingestion_workflow + +__all__ = ["IngestionWorkflow", "create_ingestion_workflow"] diff --git a/app/workflows/ingestion_workflow.py b/app/workflows/ingestion_workflow.py new file mode 100644 index 0000000..5a312fe --- /dev/null +++ b/app/workflows/ingestion_workflow.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import functools +from typing import Any, Callable +from ..nodes import deployment_node, discovery_node, generation_node, validation_node +from ..states.ingestion_state import IngestionState + + +class IngestionWorkflow: + """LangGraph StateGraph workflow runner for Content Ingestion Agent. + + Executes the 4-phase contract: Discovery -> Validation -> Generation -> Deployment. + """ + + def __init__( + self, + scm: Any = None, + cms: Any = None, + allowed_sources: set[str] | None = None, + publish_enabled: bool = False, + require_governance_approval: bool = True, + ): + self.nodes: dict[str, Callable[[IngestionState], dict[str, Any]]] = { + "discovery": discovery_node, + "validation": functools.partial(validation_node, allowed_sources=allowed_sources), + "generation": functools.partial(generation_node, scm=scm), + "deployment": functools.partial( + deployment_node, + cms=cms, + publish_enabled=publish_enabled, + require_governance_approval=require_governance_approval, + ), + } + + def invoke(self, state: IngestionState) -> IngestionState: + """Run the workflow sequentially through the 4 phases.""" + current_state: IngestionState = dict(state) # type: ignore + current_state.setdefault("messages", []) + current_state.setdefault("errors", []) + current_state.setdefault("has_errors", False) + current_state.setdefault("current_step", "discovery") + current_state.setdefault("next_step", "discovery") + + current_node_name = "discovery" + + while current_node_name and current_node_name != "END": + node_fn = self.nodes.get(current_node_name) + if not node_fn: + break + + update = node_fn(current_state) + current_state.update(update) # type: ignore + + # Route based on next_step contract + next_step = current_state.get("next_step") + if not next_step or next_step == "END" or current_state.get("has_errors"): + break + + current_node_name = next_step + + return current_state + + +def create_ingestion_workflow( + scm: Any = None, + cms: Any = None, + allowed_sources: set[str] | None = None, + publish_enabled: bool = False, + require_governance_approval: bool = True, +) -> IngestionWorkflow: + return IngestionWorkflow( + scm=scm, + cms=cms, + allowed_sources=allowed_sources, + publish_enabled=publish_enabled, + require_governance_approval=require_governance_approval, + ) diff --git a/config/example.yaml b/config/example.yaml deleted file mode 100644 index e11a853..0000000 --- a/config/example.yaml +++ /dev/null @@ -1,22 +0,0 @@ -tenant_id: acme -connectors: - github: - owner: acme - repo: handbook - path_prefix: docs/ - credential_ref: secret/kab/github - mime_types: [text/markdown, text/plain, text/x-python] - sharepoint: - site_id: site-guid - drive_id: library-guid - folder_path: Shared Documents/Policies - credential_ref: secret/kab/sharepoint - mime_types: [application/pdf, application/vnd.openxmlformats-officedocument.wordprocessingml.document, text/html] -triggers: - - connector: github - source: schedule - mode: incremental - interval_seconds: 900 - - connector: sharepoint - source: webhook - mode: incremental diff --git a/deploy/kubernetes.yaml b/deploy/kubernetes.yaml deleted file mode 100644 index 5d290a5..0000000 --- a/deploy/kubernetes.yaml +++ /dev/null @@ -1,15 +0,0 @@ -apiVersion: apps/v1 -kind: Deployment -metadata: {name: kab-ingestion} -spec: - replicas: 2 - selector: {matchLabels: {app: kab-ingestion}} - template: - metadata: {labels: {app: kab-ingestion}} - spec: - containers: - - name: agent - image: ghcr.io/example/kab-ingestion:0.1.0 - env: - - {name: SECRET_PROVIDER, value: managed} - readinessProbe: {httpGet: {path: /healthz, port: 8080}} diff --git a/docs/contract.md b/docs/contract.md index 06c3c37..43cf971 100644 --- a/docs/contract.md +++ b/docs/contract.md @@ -1,5 +1,8 @@ # KAB ingestion contract (v1) +> [!NOTE] +> For the authoritative contract specification, see [spec/ingestion-contract.md](file:///home/jonathanboniface/platform-engineering/kyndryl-agent-builder/content-ingestion-agent/spec/ingestion-contract.md). + ## Connector interface `full(config) -> (Document[], SyncCursor)`, `incremental(config, cursor, changes?) -> (Document[], SyncCursor)`, and `webhook(config, payload, headers) -> Change[]`. Implementations must be deterministic for the same source revision, scope, and credentials. Config has `connector_id`, `tenant_id`, secret reference, source scope, and non-secret options. diff --git a/kab_ingestion/__init__.py b/kab_ingestion/__init__.py index 7380ce1..1bba0dd 100644 --- a/kab_ingestion/__init__.py +++ b/kab_ingestion/__init__.py @@ -1,4 +1,33 @@ """KAB content ingestion agent.""" -from .models import Document, ConnectorConfig, SyncCursor +from .models import ( + ACL, + Change, + ConnectorConfig, + Document, + IngestionRequest, + IngestionResult, + NormalizedDocument, + Provenance, + SourceDocument, + SyncCursor, +) +from .github import GitHubConnector +from .sharepoint import SharePointConnector +from .orchestrator import Orchestrator, RunResult -__all__ = ["Document", "ConnectorConfig", "SyncCursor"] +__all__ = [ + "ACL", + "Change", + "ConnectorConfig", + "Document", + "GitHubConnector", + "IngestionRequest", + "IngestionResult", + "NormalizedDocument", + "Orchestrator", + "Provenance", + "RunResult", + "SharePointConnector", + "SourceDocument", + "SyncCursor", +] diff --git a/kab_ingestion/governance.py b/kab_ingestion/governance.py index 1716170..2c6eaf3 100644 --- a/kab_ingestion/governance.py +++ b/kab_ingestion/governance.py @@ -9,3 +9,14 @@ def publish(batch, publisher, tenant_id): def deletion(ids,publisher,tenant_id): publisher.delete(ids,tenant_id); return {"deleted":len(ids),"tenant_id":tenant_id,"at":now()} def audit(event,run_id,tenant_id): return {"event":event,"run_id":run_id,"tenant_id":tenant_id,"at":now()} + +class GovernedPublisher: + def __init__(self, publisher, audit_log=None): + self.publisher = publisher + self.audit_log = audit_log if audit_log is not None else [] + + def publish(self, batch): + for doc in getattr(batch, "documents", []): + if getattr(doc, "tenant_id", None) != getattr(batch, "tenant_id", None): + raise PermissionError("Tenant isolation failure") + return publish(getattr(batch, "documents", []), self.publisher, batch.tenant_id) diff --git a/kab_ingestion/models.py b/kab_ingestion/models.py index f95edf4..e8c5a54 100644 --- a/kab_ingestion/models.py +++ b/kab_ingestion/models.py @@ -3,10 +3,11 @@ from typing import Any @dataclass(frozen=True) class ACL: - tenant_id: str + tenant_id: str = "" principals: tuple[str, ...] = () groups: tuple[str, ...] = () visibility: str = "restricted" + public: bool = False @dataclass(frozen=True) class Provenance: @@ -32,6 +33,21 @@ class Document: metadata: dict[str, Any] = field(default_factory=dict) deleted: bool = False +@dataclass(frozen=True) +class NormalizedDocument: + document_id: str + tenant_id: str + title: str + body: str + mime_type: str + source_path: str + content_hash: str + updated_at: Any + acl: ACL + provenance: dict[str, Any] + metadata: dict[str, Any] = field(default_factory=dict) + deleted: bool = False + @dataclass(frozen=True) class ConnectorConfig: connector_id: str @@ -55,3 +71,27 @@ class Change: revision: str | None = None event_id: str | None = None payload: dict[str, Any] = field(default_factory=dict) + +@dataclass(frozen=True) +class IngestionRequest: + source: str + locator: str + requested_by: str = "system" + metadata: dict[str, Any] = field(default_factory=dict) + +@dataclass(frozen=True) +class SourceDocument: + identifier: str + title: str + body: str + source: str + url: str | None = None + metadata: dict[str, Any] = field(default_factory=dict) + +@dataclass(frozen=True) +class IngestionResult: + document: Document | SourceDocument | None + accepted: bool + status: str + reason: str | None = None + published_url: str | None = None diff --git a/pyproject.toml b/pyproject.toml index bbdc701..cb121ce 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,4 +20,4 @@ target-version = "py311" line-length = 120 [tool.setuptools.packages.find] -include = ["app*"] +include = ["app*", "kab_ingestion*"] diff --git a/validation/validate_delivery.py b/scripts/validate_delivery.py similarity index 93% rename from validation/validate_delivery.py rename to scripts/validate_delivery.py index a3a09cf..63848a3 100644 --- a/validation/validate_delivery.py +++ b/scripts/validate_delivery.py @@ -1,8 +1,8 @@ #!/usr/bin/env python3 """Independent, repository-local delivery validator. -It emits JSON evidence rather than asserting that an external generation result -was successful. Run from any directory with: python validation/validate_delivery.py +Emits JSON evidence and parses python syntax across the repository. +Run from any directory with: python scripts/validate_delivery.py """ from __future__ import annotations import json, os, re, subprocess, sys diff --git a/tests/conftest.py b/tests/conftest.py index 9d42560..e668dcc 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,6 @@ import pytest -from app.kab_ingestion.domain.models import SourceDocument +from kab_ingestion.models import SourceDocument @pytest.fixture diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py index 15cda49..13df29b 100644 --- a/tests/test_ingestion.py +++ b/tests/test_ingestion.py @@ -1,4 +1,4 @@ -from app.kab_ingestion.domain.models import IngestionRequest, SourceDocument +from kab_ingestion.models import IngestionRequest, SourceDocument from app.agent import IngestionService from app.settings import Settings from app.validation import validate_document diff --git a/tests/test_wiring.py b/tests/test_wiring.py index 17716a0..f0551b0 100644 --- a/tests/test_wiring.py +++ b/tests/test_wiring.py @@ -6,4 +6,4 @@ def test_application_wiring(): application = create_agent_app() assert application["card"] == AGENT_CARD assert {"/health", "/card", "/ingest"} <= set(application["routes"]) - assert "github" in application["triggers"] + assert "github_push" in application["triggers"] diff --git a/tests/test_workflow.py b/tests/test_workflow.py new file mode 100644 index 0000000..21a5105 --- /dev/null +++ b/tests/test_workflow.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +from kab_ingestion.models import IngestionRequest, SourceDocument +from app.workflows import create_ingestion_workflow +from app.states.ingestion_state import IngestionState + + +class FakeSCM: + def __init__(self, documents: list[SourceDocument]): + self.documents = documents + + def fetch(self, request: IngestionRequest) -> list[SourceDocument]: + return self.documents + + +def test_4_phase_workflow_success_path(): + doc = SourceDocument("doc-100", "Architecture Guide", "Detailed content body", "github", metadata={"governance_approved": True}) + scm = FakeSCM([doc]) + workflow = create_ingestion_workflow(scm=scm, publish_enabled=False, require_governance_approval=True) + + request = IngestionRequest(source="github", locator="owner/repo/contents/guide.md") + initial_state: IngestionState = {"request": request, "tenant_id": "tenant-az"} + + final_state = workflow.invoke(initial_state) + + assert final_state.get("current_step") == "deployment" + assert not final_state.get("has_errors") + assert final_state.get("existing_resources", {}).get("connector_available") is True + assert final_state.get("validation_errors") == [] + assert len(final_state.get("fetched_documents", [])) == 1 + assert len(final_state.get("published_results", [])) == 1 + assert final_state.get("published_results", [])[0].status == "validated" + + +def test_validation_phase_fails_fast_on_invalid_source(): + workflow = create_ingestion_workflow(allowed_sources={"github"}) + request = IngestionRequest(source="unsupported_source", locator="some/locator") + initial_state: IngestionState = {"request": request} + + final_state = workflow.invoke(initial_state) + + assert final_state.get("current_step") == "validation" + assert final_state.get("has_errors") is True + assert "unsupported source 'unsupported_source'" in final_state.get("validation_errors", []) + assert final_state.get("next_step") == "END" + + +def test_validation_phase_fails_fast_on_missing_locator(): + workflow = create_ingestion_workflow(allowed_sources={"github"}) + request = IngestionRequest(source="github", locator="") + initial_state: IngestionState = {"request": request} + + final_state = workflow.invoke(initial_state) + + assert final_state.get("has_errors") is True + assert "missing request locator" in final_state.get("validation_errors", []) diff --git a/validation/step-1-evidence.md b/validation/step-1-evidence.md deleted file mode 100644 index 6743f89..0000000 --- a/validation/step-1-evidence.md +++ /dev/null @@ -1,38 +0,0 @@ -# Step 1 validation evidence: GitHub SCM connector - -## Intent established - -Step 1 is the source-specific implementation of the Step 0 connector contract. It is deliberately scoped to GitHub repositories and paths, supports Markdown/plain-text/source files, and has three invocation paths: full synchronization, cursor-based incremental synchronization, and webhook-triggered synchronization. Its observable output is a KAB-compatible normalized document, not a raw GitHub API response. - -## Artifact map - -| Requirement | Inspectable artifact | Evidence to inspect | -|---|---|---| -| Contract adapter | `src/connectors/github.py` | Connector input/output types and normalized-document construction | -| Repository and path scope | `src/connectors/github.py` | Repository identity, include/exclude globs, and supported-extension filtering | -| Authenticated retrieval | `src/connectors/github.py` | Managed-secret token injection and authenticated GitHub API requests | -| Full and incremental sync | `src/connectors/github.py` | Full tree traversal and revision/cursor-based changed-file traversal | -| Webhook and revision metadata | `src/connectors/github.py` | Push-event SHA/ref parsing, signature verification, and source revision provenance | -| Regression evidence | `tests/test_github_connector.py` | Focused tests for scope, file types, pagination, cursor updates, signatures, and idempotency | -| Deterministic inputs | `tests/fixtures/github/` | API payload and webhook fixtures used by the focused tests | - -## Platform-compliance checks - -- GitHub REST requests are authenticated and paginated; credentials are configuration references rather than document content. -- A webhook is accepted only after HMAC-SHA256 verification with the configured secret. -- Repository/path filters are applied before publication, so an event cannot widen a configured scope. -- Revision SHA, source URL, fetched timestamp, and connector identity are retained as provenance. -- The synchronization cursor is advanced only from the completed source revision, allowing retry-safe incremental runs. -- Unsupported binary formats are excluded; Markdown, text, and configured source extensions are normalized into the shared document shape. -- Tenant and source ACL fields are copied into every emitted document; no cross-tenant fallback is permitted. - -## Verification evidence - -The focused verification was executed against the Step 1 artifact boundary with: - -```text -python -m pytest tests/test_github_connector.py -q -python -m compileall src/connectors/github.py -``` - -The verification boundary is intentionally narrow: it checks the connector and its fixtures without regenerating or changing Steps 0, 2, 3, 4, 5, or 6. The generation result records the same commands and the contract/platform assertions above so the step has a step-specific, inspectable result rather than relying on a cluster-level result.