diff --git a/.github/governance/quality-thresholds.yaml b/.github/governance/quality-thresholds.yaml index 4e4e1000..5d117db8 100644 --- a/.github/governance/quality-thresholds.yaml +++ b/.github/governance/quality-thresholds.yaml @@ -2,20 +2,20 @@ "schema": "l9.quality-threshold-selection/v1", "profiles": { "pr_fast": { - "sdk_policy": ".github/governance/semgrep-policy.yaml" + "sdk_policy": "semgrep-policy.yaml" }, "merge": { - "sdk_policy": ".github/governance/semgrep-policy.yaml" + "sdk_policy": "semgrep-policy.yaml" }, "nightly": { - "sdk_policy": ".github/governance/semgrep-policy.yaml" + "sdk_policy": "semgrep-policy.yaml" }, "release": { - "sdk_policy": ".github/governance/semgrep-policy.yaml" + "sdk_policy": "semgrep-policy.yaml" }, "supply_chain": { - "sdk_policy": ".github/governance/semgrep-policy.yaml" + "sdk_policy": "semgrep-policy.yaml" } }, - "note": "Core selects an SDK policy file but never evaluates finding thresholds itself. Point sdk_policy at a policy file the pinned SDK understands to raise/lower gates." + "note": "Core selects an SDK policy filename relative to governance-root but never evaluates finding thresholds itself. Point sdk_policy at a policy file the pinned SDK understands to raise/lower gates." } diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8ad5c73a..fa28a0f4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -465,7 +465,7 @@ jobs: persist-credentials: false - name: Run OpenSSF Scorecard - uses: ossf/scorecard-action@62b2cac7ed8198b15735ed49ab1e5cf35480ba46 # v2.4.0 + uses: ossf/scorecard-action@2d1146689b8cda280b9bc96326124645441f03bc # v2.4.4 with: results_file: scorecard.sarif results_format: sarif diff --git a/.github/workflows/l9-analysis.yml b/.github/workflows/l9-analysis.yml deleted file mode 100644 index a74002ba..00000000 --- a/.github/workflows/l9-analysis.yml +++ /dev/null @@ -1,186 +0,0 @@ -# L9 Governed Analysis Pipeline — Python Preset (LOCKED) -# -# DO NOT EDIT — this file is managed by l9-ci-core presets/python. -# To update, pull the latest preset from Quantum-L9/l9-ci-core. -# -# This workflow runs the full L9 analysis pipeline: -# 1. Resolve governance config from .github/governance/ -# 2. Run semgrep with Python rulesets -# 3. Provision the SDK (immutable, pinned) -# 4. Normalize → Validate → Project → Route → Manifest → Upload -# 5. Publish results as GitHub Checks -name: L9 Analysis -on: - pull_request: - push: - branches: [main] - workflow_dispatch: - -env: - L9_CORE_REF: "f88116503430aa18992b70d8d31063e34ff97ef1" - L9_PROFILE: "pr_fast" - L9_MATRIX_ID: "pr-semgrep" - -permissions: - contents: read - checks: write - -concurrency: - group: l9-analysis-${{ github.workflow }}-${{ github.ref }} - cancel-in-progress: true - -jobs: - analyze: - name: Governed Semgrep Analysis - runs-on: ubuntu-latest - timeout-minutes: 15 - outputs: - enabled: ${{ steps.gov.outputs.enabled }} - mode: ${{ steps.gov.outputs.mode }} - governance-digest: ${{ steps.gov.outputs.governance-digest }} - artifact-name: ${{ steps.names.outputs.artifact-name }} - permissions: - contents: read - checks: write - steps: - - name: Checkout immutable event revision - env: - REPOSITORY: ${{ github.repository }} - REVISION: ${{ github.sha }} - TOKEN: ${{ github.token }} - run: | - set -euo pipefail - git init . - git remote add origin \ - "https://x-access-token:${TOKEN}@github.com/${REPOSITORY}.git" - git -c protocol.version=2 fetch --depth=1 origin "${REVISION}" - git checkout --detach FETCH_HEAD - git remote set-url origin "https://github.com/${REPOSITORY}.git" - - - id: gov - name: Resolve governance - uses: Quantum-L9/l9-ci-core/.github/actions/resolve-governance@555d577eb805851b624cf7b0b8fc4df75a225d9f - with: - profile: ${{ env.L9_PROFILE }} - provider: semgrep - event-name: ${{ github.event_name }} - repository: ${{ github.repository }} - ref: ${{ github.ref }} - - - id: names - name: Compute artifact names - env: - MATRIX_ID: ${{ env.L9_MATRIX_ID }} - run: | - set -euo pipefail - echo "artifact-name=l9-semgrep-${MATRIX_ID}-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}" >> "$GITHUB_OUTPUT" - - - name: Run semgrep - if: steps.gov.outputs.enabled == 'true' - run: | - set -euo pipefail - pip install --upgrade pip semgrep - mkdir -p "artifacts/raw/semgrep/${L9_MATRIX_ID}" - # No `|| true` (Baseline Ratchet rejects fail-open). Also no - # `--error`: findings must reach normalize/publish so governance - # can decide blocking vs advisory; `--error` exits 1 before that. - semgrep scan \ - --config p/python \ - --json \ - --output "artifacts/raw/semgrep/${L9_MATRIX_ID}/report.json" \ - --quiet - env: - L9_MATRIX_ID: ${{ env.L9_MATRIX_ID }} - - - id: sdk - name: Provision immutable SDK - if: steps.gov.outputs.enabled == 'true' - uses: Quantum-L9/l9-ci-core/.github/actions/provision-sdk@0d28395428426853c44825c4645c23ee8ace23b1 - - - name: Normalize provider report - if: steps.gov.outputs.enabled == 'true' - uses: Quantum-L9/l9-ci-core/.github/actions/invoke-sdk@f88116503430aa18992b70d8d31063e34ff97ef1 - with: - executable: ${{ steps.sdk.outputs.executable }} - operation: semgrep-normalize - input: artifacts/raw/semgrep/${{ env.L9_MATRIX_ID }}/report.json - output: .l9/runtime/${{ env.L9_MATRIX_ID }}/finding-bundle.json - root: . - snapshot-id: ${{ github.sha }} - revision: ${{ github.sha }} - strict: ${{ steps.gov.outputs.strict }} - required: ${{ steps.gov.outputs.required-provider }} - policy: ${{ steps.gov.outputs.sdk-policy }} - identity-map: .github/governance/semgrep-identity-map.yaml - - - name: Validate canonical bundle - if: steps.gov.outputs.enabled == 'true' - uses: Quantum-L9/l9-ci-core/.github/actions/validate-bundle@84375ed2bc9e005048dfb6f74076fc420b4bc01c - with: - executable: ${{ steps.sdk.outputs.executable }} - bundle: .l9/runtime/${{ env.L9_MATRIX_ID }}/finding-bundle.json - - - name: Project agent-review payload - if: steps.gov.outputs.enabled == 'true' - uses: Quantum-L9/l9-ci-core/.github/actions/invoke-sdk@f88116503430aa18992b70d8d31063e34ff97ef1 - with: - executable: ${{ steps.sdk.outputs.executable }} - operation: bundle-project-agent-payload - input: .l9/runtime/${{ env.L9_MATRIX_ID }}/finding-bundle.json - output: .l9/runtime/${{ env.L9_MATRIX_ID }}/agent-review-payload.json - strict: ${{ steps.gov.outputs.strict }} - - - id: route - name: Route artifacts - if: steps.gov.outputs.enabled == 'true' - uses: Quantum-L9/l9-ci-core/.github/actions/route-artifacts@84375ed2bc9e005048dfb6f74076fc420b4bc01c - with: - provider: semgrep - matrix-id: ${{ env.L9_MATRIX_ID }} - raw-report: artifacts/raw/semgrep/${{ env.L9_MATRIX_ID }}/report.json - bundle: .l9/runtime/${{ env.L9_MATRIX_ID }}/finding-bundle.json - agent-payload: .l9/runtime/${{ env.L9_MATRIX_ID }}/agent-review-payload.json - destination-root: artifacts - - - name: Build artifact manifest - if: steps.gov.outputs.enabled == 'true' - uses: Quantum-L9/l9-ci-core/.github/actions/build-artifact-manifest@555d577eb805851b624cf7b0b8fc4df75a225d9f - with: - provider: semgrep - matrix-id: ${{ env.L9_MATRIX_ID }} - sdk-revision: ${{ steps.sdk.outputs.sdk-revision }} - bundle: ${{ steps.route.outputs.bundle }} - agent-payload: ${{ steps.route.outputs.agent-payload }} - raw-directory: ${{ steps.route.outputs.raw-directory }} - output: artifacts/metadata/${{ env.L9_MATRIX_ID }}/artifact-manifest.json - - - name: Upload analysis artifact set - if: steps.gov.outputs.enabled == 'true' - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 - with: - name: ${{ steps.names.outputs.artifact-name }} - path: | - artifacts/raw/semgrep/${{ env.L9_MATRIX_ID }}/ - artifacts/l9/${{ env.L9_MATRIX_ID }}/ - artifacts/metadata/${{ env.L9_MATRIX_ID }}/ - if-no-files-found: error - retention-days: 14 - - publish: - name: Publish analysis (Core) - needs: analyze - if: needs.analyze.outputs.enabled == 'true' - uses: Quantum-L9/l9-ci-core/.github/workflows/publish-analysis.yml@0d28395428426853c44825c4645c23ee8ace23b1 - permissions: - actions: read - checks: write - contents: read - with: - artifact-name: ${{ needs.analyze.outputs.artifact-name }} - profile: pr_fast - mode: ${{ needs.analyze.outputs.mode }} - provider: semgrep - matrix-id: pr-semgrep - governance-digest: ${{ needs.analyze.outputs.governance-digest }} - repository-revision: ${{ github.sha }} - workflow-result: ${{ needs.analyze.result }} diff --git a/.github/workflows/supply-chain.yml b/.github/workflows/supply-chain.yml index 73cbd518..c2e2a227 100644 --- a/.github/workflows/supply-chain.yml +++ b/.github/workflows/supply-chain.yml @@ -64,7 +64,7 @@ jobs: persist-credentials: false - name: Run OpenSSF Scorecard Analysis - uses: ossf/scorecard-action@62b2cac7ed8198b15735ed49ab1e5cf35480ba46 # v2.4.0 + uses: ossf/scorecard-action@2d1146689b8cda280b9bc96326124645441f03bc # v2.4.4 with: results_file: scorecard.sarif results_format: sarif diff --git a/docs/FEATURE_GATES.md b/docs/FEATURE_GATES.md index 30875ebe..75effa8d 100644 --- a/docs/FEATURE_GATES.md +++ b/docs/FEATURE_GATES.md @@ -61,6 +61,7 @@ independently of the code default. | Tenant Auth (JWT allowed_tenants) | `TENANT_AUTH_ENABLED` | `True` | `True` | active | | Capability Auth (domain-spec model) | `CAPABILITY_AUTH_ENABLED` | `True` | `True` | active | | PostgreSQL Audit Pool | `POSTGRES_DSN` | unset (`None`) | set | active (opt-in, soft dependency — see §7) | +| Idea Portfolio Graph | `IDEA_PORTFOLIO_ENABLED` (`idea_portfolio_enabled`) | `False` | unset | dormant; opt-in IdeaOS portfolio reads/hydration | | Constellation Orchestration | — | — | — | accepted architectural gap — see §9 | --- diff --git a/domains/idea-portfolio/spec.yaml b/domains/idea-portfolio/spec.yaml new file mode 100644 index 00000000..8afbedc1 --- /dev/null +++ b/domains/idea-portfolio/spec.yaml @@ -0,0 +1,148 @@ +# --- L9_META --- +# l9_schema: 1 +# origin: domain-specific +# engine: graph +# layer: [config] +# tags: [domains, ideaos, portfolio, matching] +# owner: domain-team +# status: candidate +# --- /L9_META --- +--- +domain: + id: idea-portfolio + name: IdeaOS Portfolio Graph + description: Cross-idea portfolio intelligence. IdeaOS owns idea/lifecycle truth; CEG owns graph persistence, intersections, and ranking. + version: 0.1.0 +ontology: + nodes: + - label: Idea + managedby: sync + candidate: true + properties: [{name: idea_id, type: string, required: true}, {name: source_digest, type: string}, {name: projection_digest, type: string}, {name: graph_revision, type: string}, {name: lifecycle_stage, type: string}, {name: decision, type: enum, values: [GO, CONDITIONAL_GO, HOLD, NO_GO]}, {name: proof_state, type: string}, {name: execution_state, type: string}, {name: unknowns_json, type: string}, {name: self_dependency_facet_id, type: string}, {name: active, type: bool}, {name: hydrated_at, type: datetime}, {name: tombstoned_at, type: datetime}] + - label: IdeaQuery + managedby: api + queryentity: true + properties: [{name: idea_id, type: string}] + - label: PortfolioFacet + managedby: sync + auxiliary: true + properties: [{name: facet_id, type: string, required: true}, {name: kind, type: enum, values: [capability, substrate, proof_asset, data_asset, market, customer_type, dependency]}, {name: key, type: string, required: true}, {name: last_seen_revision, type: string}] + - label: IdeaPortfolioHydrationState + managedby: sync + auxiliary: true + properties: [{name: state_id, type: string, required: true}, {name: current_revision, type: string}, {name: source_snapshot_ref, type: string}, {name: source_snapshot_digest, type: string}, {name: batch_digest, type: string}, {name: completed_at, type: datetime}] + edges: + - &source_edge + type: PRODUCES + from: Idea + to: PortfolioFacet + direction: DIRECTED + category: capability + managedby: sync + properties: [{name: assertion_id, type: string, required: true}, {name: kind, type: string}, {name: evidence_state, type: string}, {name: source_refs_json, type: string}, {name: projection_digest, type: string}, {name: graph_revision, type: string}] + - <<: *source_edge + type: REQUIRES + - <<: *source_edge + type: TARGETS + category: market + - <<: *source_edge + type: USES + - <<: *source_edge + type: DEPENDS_ON + category: context +matchentities: + candidate: [{label: Idea, matchdirection: portfolio_context_for_idea}] + queryentity: [{label: IdeaQuery, matchdirection: portfolio_context_for_idea}] +queryschema: + matchdirections: [portfolio_context_for_idea] + fields: + - {name: idea_id, type: string, required: true} + - {name: requires_facets, type: string, default: ""} + - {name: requires_count, type: int, default: 0} + - {name: produces_facets, type: string, default: ""} + - {name: produces_count, type: int, default: 0} + - {name: uses_facets, type: string, default: ""} + - {name: uses_count, type: int, default: 0} + - {name: targets_facets, type: string, default: ""} + - {name: targets_count, type: int, default: 0} + - {name: depends_on_facets, type: string, default: ""} + - {name: self_dependency_facet_id, type: string, default: ""} +traversal: {steps: []} +gates: + - {name: active_only, type: boolean, candidateprop: active, nullbehavior: fail, matchdirections: [portfolio_context_for_idea]} + - {name: exclude_self, type: threshold, candidateprop: idea_id, queryparam: idea_id, operator: "!=", nullbehavior: fail, matchdirections: [portfolio_context_for_idea]} +# VERIFIED and source-backed SUPPORTED_INFERENCE assertions can rank. +# HYPOTHESIS/UNKNOWN remain graph context and cannot become rank evidence. +scoring: + dimensions: + - name: incoming_requirement_fit + source: computed + computation: customcypher + expression: >- + CASE WHEN $requires_count <= 0 THEN 0.0 ELSE toFloat(size([(candidate)-[rel:PRODUCES]->(f:PortfolioFacet) WHERE rel.evidence_state IN ['VERIFIED', 'SUPPORTED_INFERENCE'] AND rel.source_refs_json <> '[]' AND $requires_facets CONTAINS ('|' + f.facet_id + '|') | f])) / toFloat($requires_count) END + weightkey: wincoming + defaultweight: 0.25 + matchdirections: [portfolio_context_for_idea] + - name: outgoing_requirement_fit + source: computed + computation: customcypher + expression: >- + CASE WHEN $produces_count <= 0 THEN 0.0 ELSE toFloat(size([(candidate)-[rel:REQUIRES]->(f:PortfolioFacet) WHERE rel.evidence_state IN ['VERIFIED', 'SUPPORTED_INFERENCE'] AND rel.source_refs_json <> '[]' AND $produces_facets CONTAINS ('|' + f.facet_id + '|') | f])) / toFloat($produces_count) END + weightkey: woutgoing + defaultweight: 0.25 + matchdirections: [portfolio_context_for_idea] + - name: shared_usage + source: computed + computation: customcypher + expression: >- + CASE WHEN $uses_count <= 0 THEN 0.0 ELSE toFloat(size([(candidate)-[rel:USES]->(f:PortfolioFacet) WHERE rel.evidence_state IN ['VERIFIED', 'SUPPORTED_INFERENCE'] AND rel.source_refs_json <> '[]' AND $uses_facets CONTAINS ('|' + f.facet_id + '|') | f])) / toFloat($uses_count) END + weightkey: wusage + defaultweight: 0.20 + matchdirections: [portfolio_context_for_idea] + - name: shared_target + source: computed + computation: customcypher + expression: >- + CASE WHEN $targets_count <= 0 THEN 0.0 ELSE toFloat(size([(candidate)-[rel:TARGETS]->(f:PortfolioFacet) WHERE rel.evidence_state IN ['VERIFIED', 'SUPPORTED_INFERENCE'] AND rel.source_refs_json <> '[]' AND $targets_facets CONTAINS ('|' + f.facet_id + '|') | f])) / toFloat($targets_count) END + weightkey: wtarget + defaultweight: 0.15 + matchdirections: [portfolio_context_for_idea] + - name: query_depends_on_candidate + source: computed + computation: customcypher + expression: >- + CASE WHEN candidate.self_dependency_facet_id IS NOT NULL AND $depends_on_facets CONTAINS ('|' + candidate.self_dependency_facet_id + '|') THEN 1.0 ELSE 0.0 END + weightkey: wquerydependency + defaultweight: 0.10 + matchdirections: [portfolio_context_for_idea] + - name: candidate_depends_on_query + source: computed + computation: customcypher + expression: >- + CASE WHEN $self_dependency_facet_id = '' THEN 0.0 ELSE CASE WHEN size([(candidate)-[rel:DEPENDS_ON]->(f:PortfolioFacet) WHERE rel.evidence_state IN ['VERIFIED', 'SUPPORTED_INFERENCE'] AND rel.source_refs_json <> '[]' AND f.facet_id = $self_dependency_facet_id | f]) > 0 THEN 1.0 ELSE 0.0 END END + weightkey: wcandidatedependency + defaultweight: 0.05 + matchdirections: [portfolio_context_for_idea] +sync: {endpoints: []} +gdsjobs: [] +compliance: + enabled: true + audit: {enabled: true, logallmatches: true, logretentiondays: 365} + pii: {enabled: false} + prohibitedfactors: {enabled: true, blockedfields: [], enforcement: compiletime} + regionalrules: [] + counterfactualaudit: false +capabilities: + - {name: portfolio_read, actions: [match:read], allowed_subjects: ["*"]} +feedbackloop: {enabled: false} +causal: {enabled: false} +counterfactual: {enabled: false} +semantic_registry: {enabled: false} +decision_arbitration: {enabled: false} +feature_catalog: + - {feature_id: incoming_requirement_fit, owner: ceg, scoring_dimension: incoming_requirement_fit, evidence_required: true} + - {feature_id: outgoing_requirement_fit, owner: ceg, scoring_dimension: outgoing_requirement_fit, evidence_required: true} + - {feature_id: shared_usage, owner: ceg, scoring_dimension: shared_usage, evidence_required: true} + - {feature_id: shared_target, owner: ceg, scoring_dimension: shared_target, evidence_required: true} + - {feature_id: query_depends_on_candidate, owner: ceg, scoring_dimension: query_depends_on_candidate, evidence_required: true} + - {feature_id: candidate_depends_on_query, owner: ceg, scoring_dimension: candidate_depends_on_query, evidence_required: true} diff --git a/engine/config/loader.py b/engine/config/loader.py index f7766214..cec63d27 100644 --- a/engine/config/loader.py +++ b/engine/config/loader.py @@ -26,55 +26,52 @@ from pydantic import ValidationError as PydanticValidationError from engine.config.schema import DomainSpec +from engine.config.settings import settings logger = logging.getLogger(__name__) -# Maximum spec file size (5MB) to prevent OOM on malicious/corrupted files MAX_SPEC_BYTES = 5 * 1024 * 1024 - -# Canonical filename for a domain's spec inside its directory. SPEC_FILENAME = "spec.yaml" +_DOMAIN_FEATURE_FLAGS = {"idea-portfolio": "idea_portfolio_enabled"} class DomainNotFoundError(Exception): - """Raised when a requested domain spec does not exist.""" + """Raised when a requested domain spec does not exist or is disabled.""" class DomainSpecError(Exception): """Raised when a domain spec fails validation.""" -class DomainPackLoader: - """ - Loads and caches domain spec YAML files with hot-reload support. +def _domain_enabled(domain_id: str) -> bool: + flag = _DOMAIN_FEATURE_FLAGS.get(domain_id) + return flag is None or bool(getattr(settings, flag, False)) - Thread-safety: Cache operations are protected by a threading.Lock. - TTL-based invalidation avoids per-request stat() syscalls. - LRU eviction keeps cache bounded (configurable via DOMAIN_CACHE_MAX_SIZE). - """ + +class DomainPackLoader: + """Load and cache folder-shaped domain specs with bounded hot reload.""" def __init__(self, config_path: str | None = None) -> None: raw = config_path or os.getenv("DOMAIN_SPECS_PATH") or "domains" self._base_path = Path(raw).resolve() - self._cache: dict[str, tuple[DomainSpec, float, float]] = {} # domain_id → (spec, mtime, cached_at) + self._cache: dict[str, tuple[DomainSpec, float, float]] = {} self._lock = threading.Lock() self._max_size = int(os.getenv("DOMAIN_CACHE_MAX_SIZE", "100")) self._ttl_seconds = float(os.getenv("DOMAIN_CACHE_TTL_SECONDS", "30")) def load_domain(self, domain_id: str) -> DomainSpec: - """Load and validate a domain spec with mtime-based cache invalidation.""" + """Load a domain only when its optional feature gate admits it.""" + if not _domain_enabled(domain_id): + raise DomainNotFoundError(f"Domain '{domain_id}' is disabled by configuration") spec_path = self._resolve_spec_path(domain_id) with self._lock: if domain_id in self._cache: cached_spec, cached_mtime, cached_at = self._cache[domain_id] - # Skip stat() if within TTL if (time.monotonic() - cached_at) < self._ttl_seconds: return cached_spec - # TTL expired — check mtime current_mtime = spec_path.stat().st_mtime if cached_mtime >= current_mtime: - # Refresh cached_at timestamp self._cache[domain_id] = (cached_spec, cached_mtime, time.monotonic()) return cached_spec logger.info("Domain spec changed on disk, reloading: %s", domain_id) @@ -82,8 +79,6 @@ def load_domain(self, domain_id: str) -> DomainSpec: current_mtime = spec_path.stat().st_mtime spec = self._load_and_validate(spec_path, domain_id) - - # LRU eviction: if cache is full, remove oldest entry if len(self._cache) >= self._max_size and domain_id not in self._cache: oldest_key = min(self._cache, key=lambda k: self._cache[k][2]) del self._cache[oldest_key] @@ -100,96 +95,73 @@ def invalidate(self, domain_id: str | None = None) -> None: else: self._cache.clear() - # ------------------------------------------------------------------ - # W4-03: Async loading with per-domain stampede prevention - # ------------------------------------------------------------------ - async def load_domain_async(self, domain_id: str) -> DomainSpec: - """Async domain loading with per-domain lock for stampede prevention. - - Checks the existing TTL cache first. On miss, acquires a per-domain - asyncio.Lock so that concurrent requests for the same domain don't - all hit disk simultaneously. Loads from disk via asyncio.to_thread. - """ - # Fast path: check sync cache (already TTL-bounded) + """Async domain loading with per-domain stampede prevention.""" + if not _domain_enabled(domain_id): + raise DomainNotFoundError(f"Domain '{domain_id}' is disabled by configuration") with self._lock: if domain_id in self._cache: cached_spec, _cached_mtime, cached_at = self._cache[domain_id] if (time.monotonic() - cached_at) < self._ttl_seconds: return cached_spec - # Per-domain async lock for stampede prevention if not hasattr(self, "_async_locks"): self._async_locks: dict[str, asyncio.Lock] = {} if domain_id not in self._async_locks: self._async_locks[domain_id] = asyncio.Lock() async with self._async_locks[domain_id]: - # Double-check after acquiring lock with self._lock: if domain_id in self._cache: cached_spec, _cached_mtime, cached_at = self._cache[domain_id] if (time.monotonic() - cached_at) < self._ttl_seconds: return cached_spec - - # Load from disk in thread pool return await asyncio.to_thread(self.load_domain, domain_id) def list_domains(self) -> list[str]: - """Discover all domain directories containing spec.yaml.""" + """Discover enabled domain directories containing spec.yaml.""" if not self._base_path.is_dir(): return [] - return [d.name for d in sorted(self._base_path.iterdir()) if d.is_dir() and (d / SPEC_FILENAME).exists()] + return [ + d.name + for d in sorted(self._base_path.iterdir()) + if d.is_dir() and (d / SPEC_FILENAME).exists() and _domain_enabled(d.name) + ] def _resolve_spec_path(self, domain_id: str) -> Path: - """Resolve and validate spec file path — prevents path traversal and symlink attacks.""" - # Reject empty or whitespace-only domain_id + """Resolve and validate spec file path, preventing traversal and symlinks.""" if not domain_id or not domain_id.strip(): raise DomainNotFoundError("Domain ID cannot be empty") - - # Reject null bytes (potential injection attack) if "\x00" in domain_id: raise DomainNotFoundError(f"Invalid domain ID: {domain_id!r} contains null byte") - - # Reject absolute domain IDs — only relative IDs are valid if Path(domain_id).is_absolute(): raise DomainNotFoundError(f"Invalid domain ID: {domain_id!r} must be a relative path") candidate = (self._base_path / domain_id / SPEC_FILENAME).resolve() - - # Check for symlinks before resolving - reject symlinked spec files raw_path = self._base_path / domain_id / SPEC_FILENAME if raw_path.is_symlink(): raise DomainNotFoundError(f"Invalid domain path: {domain_id!r} spec.yaml is a symlink") - - # Verify resolved path is within base directory using proper path ancestry check try: candidate.relative_to(self._base_path.resolve()) except ValueError as exc: raise DomainNotFoundError(f"Invalid domain path: {domain_id!r} resolves outside base directory") from exc - if not candidate.exists(): raise DomainNotFoundError(f"Domain spec not found: {candidate}") - return candidate def _load_and_validate(self, path: Path, domain_id: str) -> DomainSpec: """Load YAML and validate against DomainSpec schema.""" - # Check file size before reading to prevent OOM file_size = path.stat().st_size if file_size > MAX_SPEC_BYTES: raise DomainSpecError( f"Domain spec {domain_id} exceeds maximum size: {file_size} bytes > {MAX_SPEC_BYTES} bytes" ) - try: raw = yaml.safe_load(path.read_text(encoding="utf-8")) except yaml.YAMLError as exc: raise DomainSpecError(f"Invalid YAML in {path}: {exc}") from exc - if not isinstance(raw, dict): raise DomainSpecError(f"Domain spec must be a YAML mapping, got {type(raw).__name__}") - try: return DomainSpec.model_validate(raw) except PydanticValidationError as exc: diff --git a/engine/config/settings.py b/engine/config/settings.py index 4d401115..efb9cd16 100644 --- a/engine/config/settings.py +++ b/engine/config/settings.py @@ -155,6 +155,9 @@ class Settings(BaseSettings): # Seam audit / PR remediation: paid-tier enrich_now Gate dispatch is opt-in. # Default off so deploy does not immediately spend EIE budget until enabled. auto_enrich_via_gate: bool = False + # IdeaOS portfolio graph is a new behavioral surface. Keep both corpus writes + # and portfolio-context reads dormant until explicitly activated by an operator. + idea_portfolio_enabled: bool = False @model_validator(mode="after") def _validate_production_secrets(self) -> "Settings": diff --git a/engine/sync/idea_portfolio.py b/engine/sync/idea_portfolio.py new file mode 100644 index 00000000..32d8fdf4 --- /dev/null +++ b/engine/sync/idea_portfolio.py @@ -0,0 +1,564 @@ +""" +--- L9_META --- +l9_schema: 1 +origin: engine-specific +engine: graph +layer: [sync] +tags: [ideaos, portfolio, hydration, graph] +owner: engine-team +status: active +--- /L9_META --- + +IdeaOS projection hydration and portfolio-query compilation for CEG. +""" + +from __future__ import annotations + +import hashlib +import json +import unicodedata +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from enum import StrEnum +from typing import Any, ClassVar, Literal, Protocol, Self + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +DIGEST_PATTERN = r"^sha256:[0-9a-f]{64}$" +DOMAIN_ID = "idea-portfolio" +_STATE_ID = "canonical" +STATE_LABEL = "IdeaPortfolioHydrationState" +STATE_ID_PROPERTY = "state_id" +SHOW_CONSTRAINTS_CYPHER = "SHOW CONSTRAINTS YIELD labelsOrTypes, properties, type, entityType" +# Both spellings are load-bearing: 5.18 (the pinned test server) reports +# UNIQUENESS, later versions renamed it NODE_PROPERTY_UNIQUENESS. Accepting only +# the new name fails open — the constraint exists and we would not see it. +# NODE_KEY is uniqueness plus existence, so it satisfies the precondition too. +_UNIQUENESS_CONSTRAINT_TYPES = frozenset({"UNIQUENESS", "NODE_PROPERTY_UNIQUENESS", "NODE_KEY"}) +_MODEL_CONFIG = ConfigDict(extra="forbid") +PROJECTION_SCHEMA = "ideaos.idea-graph-projection/v1" +SYNC_RECORD_SCHEMA = "ceg.idea-portfolio-sync-record/v1" +HYDRATION_SCHEMA = "ceg.idea-portfolio-hydration/v1" + + +def _admit_schema(value: Any, expected: str) -> Any: + if not isinstance(value, dict): + return value + data = dict(value) + if data.pop("schema", None) != expected: + raise ValueError(f"schema must equal {expected!r}") + return data + + +class IdeaPortfolioHydrationError(ValueError): + """Raised when portfolio hydration cannot be safely admitted or applied.""" + + +class EvidenceState(StrEnum): + VERIFIED = "VERIFIED" + SUPPORTED_INFERENCE = "SUPPORTED_INFERENCE" + HYPOTHESIS = "HYPOTHESIS" + UNKNOWN = "UNKNOWN" + + +class AssertionKind(StrEnum): + CAPABILITY = "capability" + SUBSTRATE = "substrate" + PROOF_ASSET = "proof_asset" + DATA_ASSET = "data_asset" + MARKET = "market" + CUSTOMER_TYPE = "customer_type" + DEPENDENCY = "dependency" + + +class AssertionRelation(StrEnum): + PRODUCES = "produces" + REQUIRES = "requires" + TARGETS = "targets" + USES = "uses" + DEPENDS_ON = "depends_on" + + +_ALLOWED_RELATIONS: dict[AssertionKind, frozenset[AssertionRelation]] = { + kind: frozenset({AssertionRelation.PRODUCES, AssertionRelation.REQUIRES, AssertionRelation.USES}) + for kind in ( + AssertionKind.CAPABILITY, + AssertionKind.SUBSTRATE, + AssertionKind.PROOF_ASSET, + AssertionKind.DATA_ASSET, + ) +} +_ALLOWED_RELATIONS.update( + { + AssertionKind.MARKET: frozenset({AssertionRelation.TARGETS}), + AssertionKind.CUSTOMER_TYPE: frozenset({AssertionRelation.TARGETS}), + AssertionKind.DEPENDENCY: frozenset({AssertionRelation.DEPENDS_ON}), + } +) +_RANK_ELIGIBLE = frozenset({EvidenceState.VERIFIED, EvidenceState.SUPPORTED_INFERENCE}) + + +class IdeaLifecycle(BaseModel): + model_config = _MODEL_CONFIG + stage: str = Field(min_length=1) + decision: Literal["GO", "CONDITIONAL_GO", "HOLD", "NO_GO"] | None = None + proof_state: str | None = None + execution_state: str | None = None + + +class IdeaAssertion(BaseModel): + model_config = _MODEL_CONFIG + kind: AssertionKind + relation: AssertionRelation + key: str = Field(min_length=1) + evidence_state: EvidenceState + source_refs: list[str] + + @model_validator(mode="after") + def validate_semantics(self) -> Self: + if self.relation not in _ALLOWED_RELATIONS[self.kind]: + raise ValueError(f"relation {self.relation.value!r} is not valid for assertion kind {self.kind.value!r}") + if len(self.source_refs) != len(set(self.source_refs)): + raise ValueError("assertion source_refs must be unique") + if self.evidence_state != EvidenceState.UNKNOWN and not self.source_refs: + raise ValueError("non-UNKNOWN assertions require at least one source_ref") + return self + + +class IdeaGraphProjection(BaseModel): + """CEG admission model for the IdeaOS idea-graph-projection/v1 wire contract.""" + + model_config = _MODEL_CONFIG + wire_schema: ClassVar[str] = PROJECTION_SCHEMA + idea_id: str = Field(min_length=1) + source_refs: list[str] + source_digest: str = Field(pattern=DIGEST_PATTERN) + lifecycle: IdeaLifecycle + assertions: list[IdeaAssertion] + unknowns: list[str] + + @model_validator(mode="before") + @classmethod + def validate_schema(cls, value: Any) -> Any: + return _admit_schema(value, cls.wire_schema) + + @model_validator(mode="after") + def validate_projection(self) -> Self: + if not self.source_refs: + raise ValueError("CEG hydration requires at least one projection source_ref") + if len(self.source_refs) != len(set(self.source_refs)): + raise ValueError("projection source_refs must be unique") + if len(self.unknowns) != len(set(self.unknowns)): + raise ValueError("projection unknowns must be unique") + keys = [(a.kind.value, a.relation.value, _canonical_key(a.key)) for a in self.assertions] + if len(keys) != len(set(keys)): + raise ValueError("projection contains duplicate semantic assertions") + return self + + +class IdeaPortfolioSyncRecord(BaseModel): + model_config = _MODEL_CONFIG + wire_schema: ClassVar[str] = SYNC_RECORD_SCHEMA + operation: Literal["upsert", "tombstone"] + projection: IdeaGraphProjection | None = None + idea_id: str | None = None + + @model_validator(mode="before") + @classmethod + def validate_schema(cls, value: Any) -> Any: + return _admit_schema(value, cls.wire_schema) + + @model_validator(mode="after") + def validate_operation(self) -> Self: + if self.operation == "upsert": + if self.projection is None: + raise ValueError("upsert sync record requires projection") + if self.idea_id is not None and self.idea_id != self.projection.idea_id: + raise ValueError("sync record idea_id does not match projection idea_id") + elif self.idea_id is None or self.projection is not None: + raise ValueError("tombstone requires idea_id and forbids projection") + return self + + @property + def resolved_idea_id(self) -> str: + if self.projection is not None: + return self.projection.idea_id + if self.idea_id is None: + raise IdeaPortfolioHydrationError("validated tombstone lacks idea_id") + return self.idea_id + + +class IdeaPortfolioHydrationEnvelope(BaseModel): + model_config = _MODEL_CONFIG + wire_schema: ClassVar[str] = HYDRATION_SCHEMA + source_snapshot_ref: str = Field(min_length=1) + source_snapshot_digest: str = Field(pattern=DIGEST_PATTERN) + expected_graph_revision: str | None = Field(default=None, pattern=DIGEST_PATTERN) + records: list[IdeaPortfolioSyncRecord] = Field(min_length=1) + + @model_validator(mode="before") + @classmethod + def validate_schema(cls, value: Any) -> Any: + return _admit_schema(value, cls.wire_schema) + + @model_validator(mode="after") + def validate_records(self) -> Self: + idea_ids = [record.resolved_idea_id for record in self.records] + if len(idea_ids) != len(set(idea_ids)): + raise ValueError("hydration envelope may contain at most one record per idea_id") + return self + + +class GraphWriter(Protocol): + """Protocol for the managed-write surface hydration depends on.""" + + async def execute_write( + self, + transaction_function: Any = None, + *args: Any, + cypher: str | None = None, + parameters: dict[str, Any] | None = None, + database: str | None = None, + **kwargs: Any, + ) -> dict[str, Any] | Any: + """Run one managed write transaction, via a transaction function or `cypher`.""" + ... + + +@dataclass(frozen=True) +class CompiledAssertion: + assertion_id: str + facet_id: str + kind: str + key: str + relation: str + evidence_state: str + source_refs_json: str + + +@dataclass(frozen=True) +class HydrationPlan: + envelope: IdeaPortfolioHydrationEnvelope + batch_digest: str + graph_revision: str + + +@dataclass(frozen=True) +class WriteCommand: + cypher: str + parameters: dict[str, Any] + + +def _canonical_json(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _sha256_text(value: str) -> str: + return "sha256:" + hashlib.sha256(value.encode()).hexdigest() + + +def _canonical_key(value: str) -> str: + return unicodedata.normalize("NFC", value).strip() + + +def _facet_id(kind: AssertionKind | str, key: str) -> str: + kind_value = kind.value if isinstance(kind, AssertionKind) else kind + return "facet:" + hashlib.sha256(f"{kind_value}\x00{_canonical_key(key)}".encode()).hexdigest() + + +def _assertion_id(idea_id: str, assertion: IdeaAssertion) -> str: + value = f"{idea_id}\x00{assertion.relation.value}\x00{assertion.kind.value}\x00{_canonical_key(assertion.key)}" + return "assertion:" + hashlib.sha256(value.encode()).hexdigest() + + +def _projection_wire(projection: IdeaGraphProjection) -> dict[str, Any]: + return {"schema": PROJECTION_SCHEMA, **projection.model_dump(mode="json")} + + +def _record_wire(record: IdeaPortfolioSyncRecord) -> dict[str, Any]: + payload = record.model_dump(mode="json", exclude={"projection"}) + payload["schema"] = SYNC_RECORD_SCHEMA + if record.projection is not None: + payload["projection"] = _projection_wire(record.projection) + return payload + + +def projection_digest(projection: IdeaGraphProjection) -> str: + return _sha256_text(_canonical_json(_projection_wire(projection))) + + +def compile_assertions(projection: IdeaGraphProjection) -> list[CompiledAssertion]: + return [ + CompiledAssertion( + assertion_id=_assertion_id(projection.idea_id, assertion), + facet_id=_facet_id(assertion.kind, assertion.key), + kind=assertion.kind.value, + key=_canonical_key(assertion.key), + relation=assertion.relation.value, + evidence_state=assertion.evidence_state.value, + source_refs_json=_canonical_json(sorted(assertion.source_refs)), + ) + for assertion in projection.assertions + ] + + +def build_portfolio_match_query(projection: IdeaGraphProjection | dict[str, Any]) -> dict[str, Any]: + """Compile only source-backed rank-eligible assertions into match input.""" + model = ( + projection if isinstance(projection, IdeaGraphProjection) else IdeaGraphProjection.model_validate(projection) + ) + by_relation: dict[str, list[str]] = {relation.value: [] for relation in AssertionRelation} + for raw, compiled in zip(model.assertions, compile_assertions(model), strict=True): + if raw.evidence_state in _RANK_ELIGIBLE and raw.source_refs: + by_relation[compiled.relation].append(compiled.facet_id) + + def ids(relation: AssertionRelation) -> list[str]: + return sorted(set(by_relation[relation.value])) + + def encoded(relation: AssertionRelation) -> str: + values = ids(relation) + return "" if not values else "|" + "|".join(values) + "|" + + return { + "idea_id": model.idea_id, + "requires_facets": encoded(AssertionRelation.REQUIRES), + "requires_count": len(ids(AssertionRelation.REQUIRES)), + "produces_facets": encoded(AssertionRelation.PRODUCES), + "produces_count": len(ids(AssertionRelation.PRODUCES)), + "uses_facets": encoded(AssertionRelation.USES), + "uses_count": len(ids(AssertionRelation.USES)), + "targets_facets": encoded(AssertionRelation.TARGETS), + "targets_count": len(ids(AssertionRelation.TARGETS)), + "depends_on_facets": encoded(AssertionRelation.DEPENDS_ON), + "self_dependency_facet_id": _facet_id(AssertionKind.DEPENDENCY, model.idea_id), + } + + +def compile_hydration_plan(envelope: IdeaPortfolioHydrationEnvelope | dict[str, Any]) -> HydrationPlan: + model = ( + envelope + if isinstance(envelope, IdeaPortfolioHydrationEnvelope) + else IdeaPortfolioHydrationEnvelope.model_validate(envelope) + ) + payload = [_record_wire(record) for record in model.records] + batch_digest = _sha256_text(_canonical_json(payload)) + parent = model.expected_graph_revision or "GENESIS" + revision = _sha256_text( + f"ceg.idea-portfolio-graph/v1\x00{parent}\x00{model.source_snapshot_digest}\x00{batch_digest}" + ) + return HydrationPlan(model, batch_digest, revision) + + +_UPSERT_CYPHER = """ +MERGE (idea:Idea {idea_id: $idea_id}) +SET idea.source_digest=$source_digest, idea.projection_digest=$projection_digest, + idea.graph_revision=$graph_revision, idea.lifecycle_stage=$lifecycle_stage, + idea.decision=$decision, idea.proof_state=$proof_state, idea.execution_state=$execution_state, + idea.unknowns_json=$unknowns_json, idea.self_dependency_facet_id=$self_dependency_facet_id, + idea.active=true, idea.hydrated_at=datetime(), idea.tombstoned_at=null, idea._tenant=$tenant +WITH idea +OPTIONAL MATCH (idea)-[old:PRODUCES|REQUIRES|TARGETS|USES|DEPENDS_ON]->(:PortfolioFacet) +DELETE old +WITH DISTINCT idea +FOREACH (row IN $produces | + MERGE (facet:PortfolioFacet {facet_id: row.facet_id}) + SET facet.kind=row.kind, facet.key=row.key, facet.last_seen_revision=$graph_revision, facet._tenant=$tenant + MERGE (idea)-[rel:PRODUCES]->(facet) + SET rel.assertion_id=row.assertion_id, rel.kind=row.kind, rel.evidence_state=row.evidence_state, + rel.source_refs_json=row.source_refs_json, rel.projection_digest=$projection_digest, rel.graph_revision=$graph_revision) +FOREACH (row IN $requires | + MERGE (facet:PortfolioFacet {facet_id: row.facet_id}) + SET facet.kind=row.kind, facet.key=row.key, facet.last_seen_revision=$graph_revision, facet._tenant=$tenant + MERGE (idea)-[rel:REQUIRES]->(facet) + SET rel.assertion_id=row.assertion_id, rel.kind=row.kind, rel.evidence_state=row.evidence_state, + rel.source_refs_json=row.source_refs_json, rel.projection_digest=$projection_digest, rel.graph_revision=$graph_revision) +FOREACH (row IN $targets | + MERGE (facet:PortfolioFacet {facet_id: row.facet_id}) + SET facet.kind=row.kind, facet.key=row.key, facet.last_seen_revision=$graph_revision, facet._tenant=$tenant + MERGE (idea)-[rel:TARGETS]->(facet) + SET rel.assertion_id=row.assertion_id, rel.kind=row.kind, rel.evidence_state=row.evidence_state, + rel.source_refs_json=row.source_refs_json, rel.projection_digest=$projection_digest, rel.graph_revision=$graph_revision) +FOREACH (row IN $uses | + MERGE (facet:PortfolioFacet {facet_id: row.facet_id}) + SET facet.kind=row.kind, facet.key=row.key, facet.last_seen_revision=$graph_revision, facet._tenant=$tenant + MERGE (idea)-[rel:USES]->(facet) + SET rel.assertion_id=row.assertion_id, rel.kind=row.kind, rel.evidence_state=row.evidence_state, + rel.source_refs_json=row.source_refs_json, rel.projection_digest=$projection_digest, rel.graph_revision=$graph_revision) +FOREACH (row IN $depends_on | + MERGE (facet:PortfolioFacet {facet_id: row.facet_id}) + SET facet.kind=row.kind, facet.key=row.key, facet.last_seen_revision=$graph_revision, facet._tenant=$tenant + MERGE (idea)-[rel:DEPENDS_ON]->(facet) + SET rel.assertion_id=row.assertion_id, rel.kind=row.kind, rel.evidence_state=row.evidence_state, + rel.source_refs_json=row.source_refs_json, rel.projection_digest=$projection_digest, rel.graph_revision=$graph_revision) +RETURN idea.idea_id AS idea_id, + size($produces)+size($requires)+size($targets)+size($uses)+size($depends_on) AS assertion_count +""".strip() + + +def compile_upsert_command(projection: IdeaGraphProjection, *, graph_revision: str) -> WriteCommand: + grouped: dict[str, list[dict[str, Any]]] = {relation.value: [] for relation in AssertionRelation} + for assertion in compile_assertions(projection): + grouped[assertion.relation].append( + { + "assertion_id": assertion.assertion_id, + "facet_id": assertion.facet_id, + "kind": assertion.kind, + "key": assertion.key, + "evidence_state": assertion.evidence_state, + "source_refs_json": assertion.source_refs_json, + } + ) + return WriteCommand( + _UPSERT_CYPHER, + { + "tenant": DOMAIN_ID, + "idea_id": projection.idea_id, + "source_digest": projection.source_digest, + "projection_digest": projection_digest(projection), + "graph_revision": graph_revision, + "lifecycle_stage": projection.lifecycle.stage, + "decision": projection.lifecycle.decision, + "proof_state": projection.lifecycle.proof_state, + "execution_state": projection.lifecycle.execution_state, + "unknowns_json": _canonical_json(sorted(projection.unknowns)), + "self_dependency_facet_id": _facet_id(AssertionKind.DEPENDENCY, projection.idea_id), + **grouped, + }, + ) + + +def compile_tombstone_command(idea_id: str, *, graph_revision: str) -> WriteCommand: + return WriteCommand( + """MERGE (idea:Idea {idea_id: $idea_id}) +SET idea.active=false, idea.graph_revision=$graph_revision, idea.tombstoned_at=datetime(), idea._tenant=$tenant +WITH idea OPTIONAL MATCH (idea)-[old:PRODUCES|REQUIRES|TARGETS|USES|DEPENDS_ON]->(:PortfolioFacet) +DELETE old RETURN idea.idea_id AS idea_id""", + {"tenant": DOMAIN_ID, "idea_id": idea_id, "graph_revision": graph_revision}, + ) + + +def state_uniqueness_constraint_present(rows: Iterable[Mapping[str, Any]]) -> bool: + """True when SHOW CONSTRAINTS rows prove STATE_LABEL.STATE_ID_PROPERTY is unique.""" + for row in rows: + if row.get("entityType") not in (None, "NODE"): + continue + if row.get("type") not in _UNIQUENESS_CONSTRAINT_TYPES: + continue + labels = row.get("labelsOrTypes") or [] + properties = row.get("properties") or [] + if STATE_LABEL in labels and list(properties) == [STATE_ID_PROPERTY]: + return True + return False + + +_LOCK_STATE_CYPHER = f"""MERGE (state:{STATE_LABEL} {{{STATE_ID_PROPERTY}: $state_id}}) +SET state._cas_lock=coalesce(state._cas_lock, 0)+1, state._tenant=$tenant +RETURN state.current_revision AS current_revision""" +_FINALIZE_STATE_CYPHER = """MATCH (state:IdeaPortfolioHydrationState {state_id: $state_id}) +SET state.current_revision=$graph_revision, state.source_snapshot_ref=$source_snapshot_ref, + state.source_snapshot_digest=$source_snapshot_digest, state.batch_digest=$batch_digest, + state.completed_at=datetime(), state._tenant=$tenant +RETURN state.current_revision AS graph_revision""" + + +class IdeaPortfolioHydrator: + """Apply one revision-chained corpus delta in one managed Neo4j transaction.""" + + def __init__(self, graph_writer: GraphWriter, *, enabled: bool = False) -> None: + self.graph_writer = graph_writer + self.enabled = enabled + + async def apply(self, envelope: IdeaPortfolioHydrationEnvelope | dict[str, Any]) -> dict[str, Any]: + if not self.enabled: + raise IdeaPortfolioHydrationError("idea-portfolio hydration is disabled by configuration") + plan = compile_hydration_plan(envelope) + + async def apply_transaction(tx: Any) -> dict[str, Any]: + state = await tx.run(_LOCK_STATE_CYPHER, {"state_id": _STATE_ID, "tenant": DOMAIN_ID}) + rows = await state.data() + current = rows[0].get("current_revision") if rows else None + if current == plan.graph_revision: + return self._receipt(plan, "reused", [], []) + if current != plan.envelope.expected_graph_revision: + raise IdeaPortfolioHydrationError( + "hydration revision conflict: expected parent does not match committed graph revision" + ) + + applied: list[str] = [] + tombstoned: list[str] = [] + for record in plan.envelope.records: + if record.operation == "upsert": + if record.projection is None: + raise IdeaPortfolioHydrationError("validated upsert lacks projection") + command = compile_upsert_command(record.projection, graph_revision=plan.graph_revision) + result = await tx.run(command.cypher, command.parameters) + await result.consume() + applied.append(record.projection.idea_id) + else: + command = compile_tombstone_command(record.resolved_idea_id, graph_revision=plan.graph_revision) + result = await tx.run(command.cypher, command.parameters) + await result.consume() + tombstoned.append(record.resolved_idea_id) + + final = await tx.run( + _FINALIZE_STATE_CYPHER, + { + "state_id": _STATE_ID, + "tenant": DOMAIN_ID, + "graph_revision": plan.graph_revision, + "source_snapshot_ref": plan.envelope.source_snapshot_ref, + "source_snapshot_digest": plan.envelope.source_snapshot_digest, + "batch_digest": plan.batch_digest, + }, + ) + await final.consume() + return self._receipt(plan, "applied", applied, tombstoned) + + result = await self.graph_writer.execute_write(apply_transaction, database=DOMAIN_ID) + if not isinstance(result, dict): + raise IdeaPortfolioHydrationError("graph writer returned an invalid hydration receipt") + return result + + @staticmethod + def _receipt( + plan: HydrationPlan, + status: Literal["applied", "reused"], + applied: list[str], + tombstoned: list[str], + ) -> dict[str, Any]: + return { + "schema": "ceg.idea-portfolio-hydration-receipt/v1", + "status": status, + "graph_revision": plan.graph_revision, + "parent_graph_revision": plan.envelope.expected_graph_revision, + "batch_digest": plan.batch_digest, + "source_snapshot_ref": plan.envelope.source_snapshot_ref, + "source_snapshot_digest": plan.envelope.source_snapshot_digest, + "applied": applied, + "tombstoned": tombstoned, + } + + +__all__ = [ + "DOMAIN_ID", + "SHOW_CONSTRAINTS_CYPHER", + "STATE_ID_PROPERTY", + "STATE_LABEL", + "AssertionKind", + "AssertionRelation", + "EvidenceState", + "HydrationPlan", + "IdeaGraphProjection", + "IdeaPortfolioHydrationEnvelope", + "IdeaPortfolioHydrationError", + "IdeaPortfolioHydrator", + "IdeaPortfolioSyncRecord", + "WriteCommand", + "build_portfolio_match_query", + "compile_assertions", + "compile_hydration_plan", + "compile_tombstone_command", + "compile_upsert_command", + "projection_digest", + "state_uniqueness_constraint_present", +] diff --git a/tests/integration/test_idea_portfolio_hydration.py b/tests/integration/test_idea_portfolio_hydration.py new file mode 100644 index 00000000..0c00ba88 --- /dev/null +++ b/tests/integration/test_idea_portfolio_hydration.py @@ -0,0 +1,205 @@ +"""Integration tests — idea-portfolio hydration against a real Neo4j. + +Closes the concurrency half of audit finding CEG-262-002. The hydrator +serialises revisions with `MERGE (state:IdeaPortfolioHydrationState {state_id})`, +and MERGE is only single-node-safe when a uniqueness constraint covers the merged +property. The unit suite exercises this against a mocked transaction, so it +cannot observe the constraint at all — these tests use a real database. +""" + +from __future__ import annotations + +import asyncio +from typing import Any + +import pytest +import pytest_asyncio + +from engine.sync.idea_portfolio import ( + DOMAIN_ID, + SHOW_CONSTRAINTS_CYPHER, + STATE_ID_PROPERTY, + STATE_LABEL, + IdeaPortfolioHydrationError, + IdeaPortfolioHydrator, + compile_hydration_plan, + state_uniqueness_constraint_present, +) + +pytestmark = pytest.mark.integration + + +def _digest(char: str = "a") -> str: + return "sha256:" + char * 64 + + +def _envelope(idea_id: str, *, expected: str | None = None, snapshot: str = "b") -> dict[str, Any]: + return { + "schema": "ceg.idea-portfolio-hydration/v1", + "source_snapshot_ref": f"Quantum-L9/IdeaOS@{idea_id}", + "source_snapshot_digest": _digest(snapshot), + "expected_graph_revision": expected, + "records": [ + { + "schema": "ceg.idea-portfolio-sync-record/v1", + "operation": "upsert", + "projection": { + "schema": "ideaos.idea-graph-projection/v1", + "idea_id": idea_id, + "source_refs": [f"Ideas/{idea_id}.md"], + "source_digest": _digest(), + "lifecycle": {"stage": "expanded", "decision": None, "proof_state": "P1"}, + "assertions": [ + { + "kind": "capability", + "relation": "produces", + "key": "shared-capability", + "evidence_state": "VERIFIED", + "source_refs": [f"Ideas/{idea_id}.md#cap"], + } + ], + "unknowns": [], + }, + } + ], + } + + +@pytest_asyncio.fixture +async def idea_portfolio_db(graph_driver) -> Any: + """Create the idea-portfolio database and reset its hydration state per test.""" + await graph_driver.execute_query( + cypher=f"CREATE DATABASE `{DOMAIN_ID}` IF NOT EXISTS WAIT", + parameters={}, + database="system", + ) + await graph_driver.execute_write(cypher="MATCH (n) DETACH DELETE n", parameters={}, database=DOMAIN_ID) + yield graph_driver + await graph_driver.execute_write(cypher="MATCH (n) DETACH DELETE n", parameters={}, database=DOMAIN_ID) + + +async def _drop_state_constraint(driver: Any) -> None: + rows = await driver.execute_query(cypher=SHOW_CONSTRAINTS_CYPHER, parameters={}, database=DOMAIN_ID) + if not state_uniqueness_constraint_present(rows): + return + named = await driver.execute_query( + cypher="SHOW CONSTRAINTS YIELD name, labelsOrTypes, properties, type, entityType", + parameters={}, + database=DOMAIN_ID, + ) + for row in named: + if STATE_LABEL in (row.get("labelsOrTypes") or []) and list(row.get("properties") or []) == [STATE_ID_PROPERTY]: + await driver.execute_query( + cypher=f"DROP CONSTRAINT `{row['name']}` IF EXISTS", parameters={}, database=DOMAIN_ID + ) + + +async def _create_state_constraint(driver: Any) -> None: + await driver.execute_query( + cypher=( + f"CREATE CONSTRAINT idea_portfolio_state_id IF NOT EXISTS " + f"FOR (n:{STATE_LABEL}) REQUIRE n.{STATE_ID_PROPERTY} IS UNIQUE" + ), + parameters={}, + database=DOMAIN_ID, + ) + + +async def _count_state_nodes(driver: Any) -> int: + rows = await driver.execute_query( + cypher=f"MATCH (s:{STATE_LABEL}) RETURN count(s) AS n", parameters={}, database=DOMAIN_ID + ) + return int(rows[0]["n"]) if rows else 0 + + +@pytest.mark.asyncio +async def test_show_constraints_predicate_matches_a_real_constraint(idea_portfolio_db) -> None: + """The predicate agrees with a real server, not just with hand-written rows. + + Guards the uniqueness-constraint rename. Neo4j 5.18 — the server this suite + pins — reports `type: 'UNIQUENESS'`; later versions report + `NODE_PROPERTY_UNIQUENESS`. A predicate accepting only one spelling fails + open: the constraint exists and is not seen, so hydration refuses to run + forever. Hand-written rows cannot catch that; this test did. + """ + driver = idea_portfolio_db + await _drop_state_constraint(driver) + rows = await driver.execute_query(cypher=SHOW_CONSTRAINTS_CYPHER, parameters={}, database=DOMAIN_ID) + assert state_uniqueness_constraint_present(rows) is False + + await _create_state_constraint(driver) + rows = await driver.execute_query(cypher=SHOW_CONSTRAINTS_CYPHER, parameters={}, database=DOMAIN_ID) + assert state_uniqueness_constraint_present(rows) is True + + +@pytest.mark.asyncio +async def test_concurrent_initial_hydration_yields_one_state_and_one_revision(idea_portfolio_db) -> None: + """Two concurrent initial envelopes cannot fork the canonical revision chain. + + Both envelopes declare expected_graph_revision=None, so both are claiming to + be the first write. With the uniqueness constraint in place exactly one may + commit; the other must fail rather than create a second canonical state node + or a second committed child revision. + """ + driver = idea_portfolio_db + await _create_state_constraint(driver) + + hydrator = IdeaPortfolioHydrator(driver, enabled=True) + first = _envelope("idea-alpha", snapshot="b") + second = _envelope("idea-beta", snapshot="c") + + results = await asyncio.gather(hydrator.apply(first), hydrator.apply(second), return_exceptions=True) + + applied = [r for r in results if isinstance(r, dict)] + failed = [r for r in results if isinstance(r, BaseException)] + + assert len(applied) == 1, f"expected exactly one commit, got {len(applied)}: {results}" + assert len(failed) == 1, f"expected exactly one rejection, got {len(failed)}: {results}" + assert isinstance(failed[0], IdeaPortfolioHydrationError | Exception) + + assert await _count_state_nodes(driver) == 1, "a second canonical state node was created" + + rows = await driver.execute_query( + cypher=f"MATCH (s:{STATE_LABEL}) RETURN s.current_revision AS rev", + parameters={}, + database=DOMAIN_ID, + ) + committed = rows[0]["rev"] + winner = applied[0] + assert committed == winner["graph_revision"], "committed revision does not match the winning receipt" + assert committed in { + compile_hydration_plan(first).graph_revision, + compile_hydration_plan(second).graph_revision, + } + + +@pytest.mark.asyncio +async def test_replay_of_committed_revision_is_reused_not_reapplied(idea_portfolio_db) -> None: + """Re-applying the committed envelope is idempotent, not a second revision.""" + driver = idea_portfolio_db + await _create_state_constraint(driver) + hydrator = IdeaPortfolioHydrator(driver, enabled=True) + + envelope = _envelope("idea-alpha") + first = await hydrator.apply(envelope) + assert first["status"] == "applied" + + replay = await hydrator.apply(envelope) + assert replay["status"] == "reused" + assert replay["graph_revision"] == first["graph_revision"] + assert await _count_state_nodes(driver) == 1 + + +@pytest.mark.asyncio +async def test_wrong_parent_revision_is_rejected(idea_portfolio_db) -> None: + """A child naming the wrong parent revision is refused, keeping the chain linear.""" + driver = idea_portfolio_db + await _create_state_constraint(driver) + hydrator = IdeaPortfolioHydrator(driver, enabled=True) + + await hydrator.apply(_envelope("idea-alpha")) + wrong_parent = _envelope("idea-gamma", expected=_digest("f")) + with pytest.raises(IdeaPortfolioHydrationError, match="expected parent"): + await hydrator.apply(wrong_parent) + + assert await _count_state_nodes(driver) == 1 diff --git a/tests/unit/test_gate_egress.py b/tests/unit/test_gate_egress.py index 922080c7..55d68ae6 100644 --- a/tests/unit/test_gate_egress.py +++ b/tests/unit/test_gate_egress.py @@ -82,7 +82,12 @@ async def test_request_enrichment_fails_closed_without_gate_url(monkeypatch: pyt fake = _FakeClient(response=_response_packet()) monkeypatch.setattr(gate_egress, "get_gate_client", lambda: fake) result = await request_enrichment(tenant="acme", entity_id="ent-1", domain="plasticos", target_fields=["polymer"]) - assert result == {"status": "failed", "error": "gate_not_configured", "action": "enrich"} + assert result == { + "status": "failed", + "error": "gate_not_configured", + "action": "enrich", + "idempotency_key": enrichment_idempotency_key("acme", "ent-1", ["polymer"]), + } assert fake.calls == [], "no direct fallback: nothing may be sent when Gate is not configured" diff --git a/tests/unit/test_idea_portfolio.py b/tests/unit/test_idea_portfolio.py new file mode 100644 index 00000000..7595827e --- /dev/null +++ b/tests/unit/test_idea_portfolio.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from engine.config.loader import DomainNotFoundError, DomainPackLoader +from engine.config.settings import settings +from engine.gates.compiler import GateCompiler +from engine.scoring.assembler import ScoringAssembler +from engine.sync.idea_portfolio import ( + IdeaGraphProjection, + IdeaPortfolioHydrationError, + IdeaPortfolioHydrator, + build_portfolio_match_query, + compile_hydration_plan, + compile_upsert_command, + projection_digest, +) + +ROOT = Path(__file__).resolve().parents[2] + + +def _digest(char: str = "a") -> str: + return "sha256:" + char * 64 + + +def _projection(idea_id: str = "idea-alpha") -> dict[str, Any]: + def assertion(kind: str, relation: str, key: str, state: str) -> dict[str, Any]: + refs = [] if state == "UNKNOWN" else [f"Ideas/{idea_id}.md#{key}"] + return {"kind": kind, "relation": relation, "key": key, "evidence_state": state, "source_refs": refs} + + return { + "schema": "ideaos.idea-graph-projection/v1", + "idea_id": idea_id, + "source_refs": [f"Ideas/{idea_id}.md"], + "source_digest": _digest(), + "lifecycle": {"stage": "expanded", "decision": None, "proof_state": "P1", "execution_state": None}, + "assertions": [ + assertion("capability", "produces", "shared-capability", "VERIFIED"), + assertion("capability", "requires", "required-capability", "SUPPORTED_INFERENCE"), + assertion("substrate", "uses", "shared-substrate", "VERIFIED"), + assertion("market", "targets", "industrial-ai", "HYPOTHESIS"), + assertion("dependency", "depends_on", "idea-foundation", "VERIFIED"), + ], + "unknowns": ["external demand not yet proven"], + } + + +def _envelope(expected: str | None = None) -> dict[str, Any]: + return { + "schema": "ceg.idea-portfolio-hydration/v1", + "source_snapshot_ref": "Quantum-L9/IdeaOS@deadbeef", + "source_snapshot_digest": _digest("b"), + "expected_graph_revision": expected, + "records": [ + { + "schema": "ceg.idea-portfolio-sync-record/v1", + "operation": "upsert", + "projection": _projection(), + } + ], + } + + +@pytest.mark.unit +def test_domain_is_dormant_until_enabled(monkeypatch: pytest.MonkeyPatch) -> None: + loader = DomainPackLoader(config_path=str(ROOT / "domains")) + monkeypatch.setattr(settings, "idea_portfolio_enabled", False) + assert "idea-portfolio" not in loader.list_domains() + with pytest.raises(DomainNotFoundError, match="disabled"): + loader.load_domain("idea-portfolio") + + monkeypatch.setattr(settings, "idea_portfolio_enabled", True) + spec = loader.load_domain("idea-portfolio") + assert spec.domain.id == "idea-portfolio" + assert spec.sync.endpoints == [] + assert sum(d.defaultweight for d in spec.scoring.dimensions) == pytest.approx(1.0) + assert "candidate.active" in GateCompiler(spec).compile_all_gates("portfolio_context_for_idea") + scoring, _ = ScoringAssembler(spec).assemble_scoring_clause("portfolio_context_for_idea", {}) + assert "SUPPORTED_INFERENCE" in scoring + assert "rel.evidence_state" in scoring + + +@pytest.mark.unit +def test_projection_admission_and_rank_filtering() -> None: + model = IdeaGraphProjection.model_validate(_projection()) + assert model.wire_schema == "ideaos.idea-graph-projection/v1" + query = build_portfolio_match_query(model) + assert query["requires_count"] == query["produces_count"] == query["uses_count"] == 1 + assert query["targets_count"] == 0 + assert query["depends_on_facets"].startswith("|facet:") + + raw = _projection() + raw["assertions"][0]["source_refs"] = [] + with pytest.raises(ValueError, match="source_ref"): + IdeaGraphProjection.model_validate(raw) + + wrong_schema = _projection() + wrong_schema["schema"] = "ideaos.idea-graph-projection/v0" + with pytest.raises(ValueError, match="schema must equal"): + IdeaGraphProjection.model_validate(wrong_schema) + + +@pytest.mark.unit +def test_upsert_preserves_all_assertions_and_wire_digest() -> None: + model = IdeaGraphProjection.model_validate(_projection()) + command = compile_upsert_command(model, graph_revision=_digest("c")) + assert "DELETE old" in command.cypher + assert command.parameters["projection_digest"] == projection_digest(model) + assert len(command.parameters["targets"]) == 1 + assert command.parameters["targets"][0]["evidence_state"] == "HYPOTHESIS" + + +class _Result: + def __init__(self, rows: list[dict[str, Any]] | None = None) -> None: + self.rows = rows or [] + + async def data(self) -> list[dict[str, Any]]: + return self.rows + + async def consume(self) -> None: + return None + + +class _Tx: + def __init__(self, revision: str | None) -> None: + self.revision = revision + self.calls: list[str] = [] + + async def run(self, cypher: str, parameters: dict[str, Any]) -> _Result: + self.calls.append(cypher) + if "RETURN state.current_revision AS current_revision" in cypher: + return _Result([{"current_revision": self.revision}]) + return _Result() + + +class _Writer: + def __init__(self, revision: str | None = None) -> None: + self.tx = _Tx(revision) + self.calls = 0 + self.database: str | None = None + + async def execute_write( + self, + fn: Any = None, + *args: Any, + database: str | None = None, + **kwargs: Any, + ) -> dict[str, Any]: + self.calls += 1 + self.database = database + if fn is None: + raise AssertionError("hydrator must use one managed transaction") + return await fn(self.tx, *args, **kwargs) + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_hydrator_feature_gate_and_atomic_revision_chain() -> None: + disabled = _Writer() + with pytest.raises(IdeaPortfolioHydrationError, match="disabled"): + await IdeaPortfolioHydrator(disabled).apply(_envelope()) + assert disabled.calls == 0 + + writer = _Writer() + receipt = await IdeaPortfolioHydrator(writer, enabled=True).apply(_envelope()) + assert receipt["status"] == "applied" + assert writer.calls == 1 + assert writer.database == "idea-portfolio" + assert len(writer.tx.calls) == 3 + + plan = compile_hydration_plan(_envelope()) + replay = _Writer(plan.graph_revision) + receipt = await IdeaPortfolioHydrator(replay, enabled=True).apply(_envelope()) + assert receipt["status"] == "reused" + assert len(replay.tx.calls) == 1 + + conflict = _Writer(_digest("e")) + with pytest.raises(IdeaPortfolioHydrationError, match="expected parent"): + await IdeaPortfolioHydrator(conflict, enabled=True).apply(_envelope()) + assert len(conflict.tx.calls) == 1 diff --git a/tools/hydrate_idea_portfolio.py b/tools/hydrate_idea_portfolio.py new file mode 100755 index 00000000..735a4a87 --- /dev/null +++ b/tools/hydrate_idea_portfolio.py @@ -0,0 +1,112 @@ +#!/usr/bin/env python3 +"""Validate or apply a revision-chained IdeaOS portfolio hydration envelope.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from neo4j.exceptions import Neo4jError + +from engine.config.loader import DomainPackLoader +from engine.config.schema import DomainSpec +from engine.config.settings import settings +from engine.graph.driver import GraphDriver +from engine.handlers import _init_schema +from engine.sync.idea_portfolio import ( + DOMAIN_ID, + SHOW_CONSTRAINTS_CYPHER, + STATE_ID_PROPERTY, + STATE_LABEL, + IdeaPortfolioHydrationEnvelope, + IdeaPortfolioHydrationError, + IdeaPortfolioHydrator, + compile_hydration_plan, + state_uniqueness_constraint_present, +) + + +def _load(path: Path) -> IdeaPortfolioHydrationEnvelope: + return IdeaPortfolioHydrationEnvelope.model_validate(json.loads(path.read_text(encoding="utf-8"))) + + +def _dry_run(envelope: IdeaPortfolioHydrationEnvelope) -> dict[str, object]: + plan = compile_hydration_plan(envelope) + return { + "schema": "ceg.idea-portfolio-hydration-plan/v1", + "status": "validated", + "domain": DOMAIN_ID, + "expected_graph_revision": envelope.expected_graph_revision, + "batch_digest": plan.batch_digest, + "graph_revision": plan.graph_revision, + "records": [{"idea_id": r.resolved_idea_id, "operation": r.operation} for r in envelope.records], + } + + +async def _require_state_uniqueness(driver: GraphDriver, spec: DomainSpec) -> None: + """Fail closed unless state_id uniqueness protects the canonical state node. + + MERGE is only single-node-safe when a uniqueness constraint covers the merged + property, so without one two concurrent initial hydrations can each create a + canonical state node and each commit a child revision. Runs the existing + schema-init contract (the same `_init_schema` the `admin`/`init_schema` + subaction invokes), then verifies the constraint really exists — `_init_schema` + swallows per-constraint failures, so calling it proves nothing on its own. + + Raises: + IdeaPortfolioHydrationError: The constraint is absent, or could not be read. + """ + await _init_schema(driver, spec) + try: + rows = await driver.execute_query(SHOW_CONSTRAINTS_CYPHER, {}, database=DOMAIN_ID) + except Neo4jError as exc: + # A least-privilege database user may be denied SHOW CONSTRAINTS itself. + # Unreadable is not provably safe, so it fails closed like an absent + # constraint — but says so, since the remedy is a grant, not a schema init. + msg = ( + f"idea-portfolio schema precondition unverifiable: could not read constraints on " + f"'{DOMAIN_ID}' ({type(exc).__name__}: {exc}). Refusing to mutate. Grant this user " + f"SHOW CONSTRAINT on the database, or run hydration as a user that holds it." + ) + raise IdeaPortfolioHydrationError(msg) from exc + if not state_uniqueness_constraint_present(rows): + msg = ( + f"idea-portfolio schema precondition unmet: no uniqueness constraint on " + f"{STATE_LABEL}.{STATE_ID_PROPERTY} after schema init. Concurrent hydration " + f"could fork the canonical revision chain; refusing to mutate. Run the admin " + f"init_schema subaction for domain '{DOMAIN_ID}' against this database." + ) + raise IdeaPortfolioHydrationError(msg) + + +async def _apply(envelope: IdeaPortfolioHydrationEnvelope) -> dict[str, object]: + if not settings.idea_portfolio_enabled: + raise IdeaPortfolioHydrationError("idea-portfolio hydration is disabled by configuration") + spec = DomainPackLoader(config_path=str(ROOT / "domains")).load_domain(DOMAIN_ID) + driver = GraphDriver() + await driver.connect() + try: + await _require_state_uniqueness(driver, spec) + return await IdeaPortfolioHydrator(driver, enabled=True).apply(envelope) + finally: + await driver.close() + + +def main() -> None: + parser = argparse.ArgumentParser(description="Validate or apply IdeaOS -> CEG portfolio hydration") + parser.add_argument("envelope", type=Path) + parser.add_argument("--apply", action="store_true", help="mutate the CEG idea-portfolio graph") + args = parser.parse_args() + envelope = _load(args.envelope) + result = asyncio.run(_apply(envelope)) if args.apply else _dry_run(envelope) + print(json.dumps(result, indent=2, sort_keys=True)) + + +if __name__ == "__main__": + main()