decomposer: generate deliverable files for Define the normalized ingestion contract and pluggable source-connector interface for SCM, CMS, and other content sources.; Implement a GitHub SCM connector that conforms to the normalized ingestion contract and supports scoped full and incremental ingestion of Markdown, plain-text, and source files with webhook and revision metadata.; Implement a SharePoint CMS connector that conforms to the normalized ingestion contract and supports scoped full and incremental ingestion of PDF, DOCX, and HTML files with webhook and revision metadata.; Implement ingestion orchestration and triggers; Normalize, govern, and publish ingested content to the shared knowledge store.; Add deployment/configuration, tests, and documentation for the ingestion agent.; Validate end-to-end ingestion and downstream content availability.

kab_ingestion/orchestrator.py

@@ -1,31 +1,21 @@
-from __future__ import annotations
-import uuid
-from dataclasses import dataclass
-from .models import IngestionBatch
+from dataclasses import dataclass, field
+from .models import *
+from .ports import Connector, Publisher
+from .util import now
 
 @dataclass
 class RunResult:
-    run_id: str
-    status: str
-    batch: IngestionBatch | None = None
-    attempts: int = 0
-    dead_letter: list[dict] | None = None
+    run_id: str; status: str; count: int=0; errors: list[str]=field(default_factory=list); cursor: SyncCursor|None=None
 
 class Orchestrator:
-    def __init__(self, connectors, cursor_store, publisher, max_attempts=3):
-        self.connectors, self.cursors, self.publisher, self.max_attempts = connectors, cursor_store, publisher, max_attempts
-
-    def run(self, connector_name, tenant_id, mode='incremental', idempotency_key=None):
-        connector = self.connectors[connector_name]
-        cursor = self.cursors.load(tenant_id, connector_name)
-        batch = None
-        for attempt in range(1, self.max_attempts + 1):
+    def __init__(self, connectors: dict[str,Connector], publisher: Publisher, max_retries=3): self.connectors,self.publisher,self.max_retries=connectors,publisher,max_retries
+    def run(self, config, cursor=None, mode="incremental", changes=None, run_id=None):
+        run_id=run_id or f"{config.connector_id}:{now()}"
+        for attempt in range(self.max_retries):
             try:
-                batch = connector.full(tenant_id) if mode == 'full' else connector.incremental(tenant_id, cursor)
-                self.publisher.publish(batch)
-                if batch.next_cursor: self.cursors.save(batch.next_cursor)
-                return RunResult(str(uuid.uuid4()), 'succeeded', batch, attempt, [])
-            except Exception as exc:
-                if attempt == self.max_attempts:
-                    return RunResult(str(uuid.uuid4()), 'dead_lettered', batch, attempt, [{'error': type(exc).__name__}])
-        raise AssertionError('unreachable')
+                docs,new_cursor=(self.connectors[config.connector_id].full(config) if mode=="full" else self.connectors[config.connector_id].incremental(config,cursor or SyncCursor(config.connector_id,"incremental"),changes))
+                self.publisher.upsert(docs); return RunResult(run_id,"succeeded",len(docs),cursor=new_cursor)
+            except Exception as e:
+                if attempt==self.max_retries-1: return RunResult(run_id,"dead_letter",errors=[str(e)])
+        return RunResult(run_id,"dead_letter",errors=["unreachable"])
+    def webhook(self,config,payload,headers): return self.connectors[config.connector_id].webhook(config,payload,headers)