main to develop

.github/workflows/publish.yml

@@ -11,6 +11,7 @@
   contents: write
   id-token: write # Required for PyPI OIDC Trusted Publishing and Sigstore
   pages: write    # Required for GitHub Pages deployment
+  attestations: write # Required for SLSA build provenance
 
 env:
   FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
@@ -48,10 +49,15 @@
           output-file: sbom.spdx.json
           upload-release-assets: false
 
+      - name: Generate SLSA Provenance
+        uses: actions/attest-build-provenance@v2
+        with:
+          subject-path: |
+            dist/*.whl
+            dist/*.tar.gz
+
       - name: Publish to PyPI
         uses: pypa/gh-action-pypi-publish@release/v1
-        with:
-          attestations: false  # Requires Enterprise Cloud or public repo
 
       - name: Sign Wheel
         uses: sigstore/gh-action-sigstore-python@v3.3.0
@@ -97,3 +103,66 @@
       - name: Deploy to GitHub Pages
         id: deployment
         uses: actions/deploy-pages@v4
+
+  publish-container:
+    needs: release
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+      id-token: write
+      attestations: write
+    steps:
+      - name: Harden Runner
+        uses: step-security/harden-runner@v2
+        with:
+          egress-policy: audit
+
+      - uses: actions/checkout@v4
+
+      - name: Log in to GitHub Container Registry
+        uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ghcr.io/${{ github.repository }}
+          tags: |
+            type=semver,pattern={{version}}
+            type=semver,pattern={{major}}.{{minor}}
+            type=sha
+            type=raw,value=latest
+
+      - name: Build and push Docker image
+        id: build-and-push
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+
+      - name: Generate SBOM for Container
+        uses: anchore/sbom-action@v0
+        with:
+          image: ghcr.io/${{ github.repository }}@${{ steps.build-and-push.outputs.digest }}
+          format: spdx-json
+          output-file: sbom-container.spdx.json
+
+      - name: Install Cosign
+        uses: sigstore/cosign-installer@v3.5.0
+
+      - name: Sign the image with Cosign
+        run: cosign sign --yes ghcr.io/${{ github.repository }}@${{ steps.build-and-push.outputs.digest }}
+
+      - name: Generate SLSA Provenance for Container
+        uses: actions/attest-build-provenance@v2
+        with:
+          subject-name: ghcr.io/${{ github.repository }}
+          subject-digest: ${{ steps.build-and-push.outputs.digest }}
+          push-to-registry: true

pyproject.toml

@@ -15,9 +15,10 @@ authors = [
     { name = "Gowtham A Rao", email = "gowtham.rao@coreason.ai" },
 ]
 dependencies = [
-    "coreason-manifest>=0.61.1",
+    "coreason-manifest>=0.72.1",
     "coreason-urn-authority>=0.11.1",
     "httpx>=0.28.1",
+    "hvac>=2.4.0",
     "libcst>=1.8.6",
     "loguru>=0.7.2",
     "mcp>=1.3.0",
@@ -64,6 +65,7 @@ ignore = ["DEP002"]
 
 [tool.deptry.per_rule_ignores]
 DEP003 = ["coreason_manifest"]
+DEP001 = ["coreason_runtime"]
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/coreason_meta_engineering"]
@@ -79,7 +81,7 @@ target-version = "py314"
 
 [tool.ruff.lint]
 select = ["E", "F", "B", "I", "UP", "SIM", "RUF", "ARG", "C4", "PT", "TCH", "FA", "PIE", "RET", "PERF", "FURB", "LOG", "N", "A", "S"]
-ignore = ["S101", "TC001", "TC002", "TC003", "UP037"]
+ignore = ["S101", "TC001", "TC002", "TC003", "UP037", "E501"]
 
 [tool.ruff.lint.isort]
 known-first-party = ["coreason_meta_engineering"]

src/coreason_meta_engineering/forge_orchestrator.py

@@ -10,6 +10,7 @@
 
 import asyncio
 import json
+import os
 import typing
 from pathlib import Path
 
@@ -18,35 +19,71 @@
 from coreason_meta_engineering.pvv import execute_pvv_pipeline
 from coreason_meta_engineering.utils.logger import logger
 
+__all__ = ["DynamicForgeOrchestrator", "dispatch_agent_generation", "orchestrate_generation"]
+
 try:
-    from coreason_runtime.execution_plane.fabricator import dispatch_agent_generation
+    from coreason_runtime.execution_plane.fabricator import dispatch_agent_generation  # type: ignore
 except ImportError:
     # If not running in a full swarm, fallback logic can be placed here or we just raise.
     async def dispatch_agent_generation(prompt_context: str) -> typing.Any:
         if "actionspace:substrate:test_crd" in prompt_context:
-            return {"payload": "from pydantic import BaseModel\nfrom typing import ClassVar\nclass KubernetesCRDBase(BaseModel): pass\nclass Testcrd(KubernetesCRDBase):\n    api_group: ClassVar[str] = \"test.group\"\n    name: str\n\nTestcrd.model_rebuild()\n", "deliberation_trace": "test"}
+            return {
+                "payload": 'from pydantic import BaseModel\nfrom typing import ClassVar\nclass KubernetesCRDBase(BaseModel): pass\nclass Testcrd(KubernetesCRDBase):\n    api_group: ClassVar[str] = "test.group"\n    name: str\n\nTestcrd.model_rebuild()\n',
+                "deliberation_trace": "test",
+            }
         if "TestModelClass" in prompt_context or "Test Model Class" in prompt_context:
-            return {"payload": "from typing import Optional\nfrom pydantic import BaseModel\nclass CoreasonBaseState(BaseModel): pass\nclass TestModelClass(CoreasonBaseState):\n    name: str\n    count: Optional[int] = None\n\nTestModelClass.model_rebuild()\n", "deliberation_trace": "test"}
+            return {
+                "payload": "from typing import Optional\nfrom pydantic import BaseModel\nclass CoreasonBaseState(BaseModel): pass\nclass TestModelClass(CoreasonBaseState):\n    name: str\n    count: Optional[int] = None\n\nTestModelClass.model_rebuild()\n",
+                "deliberation_trace": "test",
+            }
         if "my_actuator" in prompt_context:
-            return {"payload": "class DummyMCP:\n    def tool(self):\n        return lambda f: f\nmcp = DummyMCP()\nfrom pydantic import BaseModel\nclass Dummy(BaseModel):\n    name: str\n    age: int\n    is_active: bool\n@mcp.tool()\ndef my_actuator_func(name: str) -> str:\n    pass\n", "deliberation_trace": "test"}
+            return {
+                "payload": "class DummyMCP:\n    def tool(self):\n        return lambda f: f\nmcp = DummyMCP()\nfrom pydantic import BaseModel\nclass Dummy(BaseModel):\n    name: str\n    age: int\n    is_active: bool\n@mcp.tool()\ndef my_actuator_func(name: str) -> str:\n    pass\n",
+                "deliberation_trace": "test",
+            }
         if "my_agent" in prompt_context:
-            return {"payload": "from pydantic import BaseModel\nclass CoreasonBaseAgent(BaseModel): pass\nclass MyAgentClass(CoreasonBaseAgent):\n    pass\n\nMyAgentClass.model_rebuild()\n", "deliberation_trace": "test"}
+            return {
+                "payload": "from pydantic import BaseModel\nclass CoreasonBaseAgent(BaseModel): pass\nclass MyAgentClass(CoreasonBaseAgent):\n    pass\n\nMyAgentClass.model_rebuild()\n",
+                "deliberation_trace": "test",
+            }
         if "Class1InvalidClassStart" in prompt_context:
-            return {"payload": "from pydantic import BaseModel\nclass CoreasonBaseState(BaseModel): pass\nclass Class1InvalidClassStart(CoreasonBaseState):\n    pass\n\nClass1InvalidClassStart.model_rebuild()\n", "deliberation_trace": "test"}
+            return {
+                "payload": "from pydantic import BaseModel\nclass CoreasonBaseState(BaseModel): pass\nclass Class1InvalidClassStart(CoreasonBaseState):\n    pass\n\nClass1InvalidClassStart.model_rebuild()\n",
+                "deliberation_trace": "test",
+            }
         if "actionspace:node:test" in prompt_context:
-            return {"payload": "from pydantic import BaseModel\nclass CoreasonBaseAgent(BaseModel): pass\nclass GeneratedClass(CoreasonBaseAgent):\n    pass\n\nGeneratedClass.model_rebuild()\n", "deliberation_trace": "test"}
+            return {
+                "payload": "from pydantic import BaseModel\nclass CoreasonBaseAgent(BaseModel): pass\nclass GeneratedClass(CoreasonBaseAgent):\n    pass\n\nGeneratedClass.model_rebuild()\n",
+                "deliberation_trace": "test",
+            }
         if "tool_1_actuator" in prompt_context:
-            return {"payload": "class DummyMCP:\n    def tool(self):\n        return lambda f: f\nmcp = DummyMCP()\nfrom pydantic import BaseModel\nclass Dummy(BaseModel): pass\n@mcp.tool()\ndef tool_1_actuator() -> str:\n    pass\n", "deliberation_trace": "test"}
-        if "generated_identifier" in prompt_context or ("actionspace:solver" in prompt_context and "___" in prompt_context):
-            return {"payload": "class DummyMCP:\n    def tool(self):\n        return lambda f: f\nmcp = DummyMCP()\nfrom pydantic import BaseModel\nclass Dummy(BaseModel): pass\n@mcp.tool()\ndef generated_identifier() -> str:\n    pass\n", "deliberation_trace": "test"}
+            return {
+                "payload": "class DummyMCP:\n    def tool(self):\n        return lambda f: f\nmcp = DummyMCP()\nfrom pydantic import BaseModel\nclass Dummy(BaseModel): pass\n@mcp.tool()\ndef tool_1_actuator() -> str:\n    pass\n",
+                "deliberation_trace": "test",
+            }
+        if "generated_identifier" in prompt_context or (
+            "actionspace:solver" in prompt_context and "___" in prompt_context
+        ):
+            return {
+                "payload": "class DummyMCP:\n    def tool(self):\n        return lambda f: f\nmcp = DummyMCP()\nfrom pydantic import BaseModel\nclass Dummy(BaseModel): pass\n@mcp.tool()\ndef generated_identifier() -> str:\n    pass\n",
+                "deliberation_trace": "test",
+            }
         if "DummyState" in prompt_context or "Dummystate" in prompt_context:
-            return {"payload": "from typing import Annotated\nfrom pydantic import BaseModel\nclass CoreasonBaseState(BaseModel): pass\nclass DummyState(CoreasonBaseState):\n    name: Annotated[str, 'test']\n\nDummyState.model_rebuild()\n", "deliberation_trace": "test"}
+            return {
+                "payload": "from typing import Annotated\nfrom pydantic import BaseModel\nclass CoreasonBaseState(BaseModel): pass\nclass DummyState(CoreasonBaseState):\n    name: Annotated[str, 'test']\n\nDummyState.model_rebuild()\n",
+                "deliberation_trace": "test",
+            }
 
         # Default fallback for any other tests
         if "actionspace:solver" in prompt_context:
-            return {"payload": "from typing import Optional\nfrom pydantic import BaseModel\nclass CoreasonBaseState(BaseModel): pass\nclass TestModelClass(CoreasonBaseState):\n    name: str\n    count: Optional[int] = None\n\nTestModelClass.model_rebuild()\n", "deliberation_trace": "test"}
-
-        raise NotImplementedError(f"Dynamic forge requires coreason_runtime.execution_plane.fabricator. Prompt was: {prompt_context[:100]}")
+            return {
+                "payload": "from typing import Optional\nfrom pydantic import BaseModel\nclass CoreasonBaseState(BaseModel): pass\nclass TestModelClass(CoreasonBaseState):\n    name: str\n    count: Optional[int] = None\n\nTestModelClass.model_rebuild()\n",
+                "deliberation_trace": "test",
+            }
+
+        raise NotImplementedError(
+            f"Dynamic forge requires coreason_runtime.execution_plane.fabricator. Prompt was: {prompt_context[:100]}"
+        )
 
 
 class DynamicForgeOrchestrator:
@@ -121,6 +158,22 @@ async def scaffold_ast(
         target_file = Path(target_file_path)
         if target_file.is_dir():
             raise ValueError(f"Target path {target_file} is a directory, not a file.")
+
+        # --- License Chronometer: AST Guillotine ---
+        if os.environ.get("AST_GUILLOTINE_ACTIVE") == "True":
+            license_header = (
+                "# Copyright (c) 2026 CoReason, Inc\n"
+                "#\n"
+                "# This software is proprietary and dual-licensed\n"
+                '# Licensed under the Prosperity Public License 3.0 (the "License")\n'
+                "# A copy of the license is available at <https://prosperitylicense.com/versions/3.0.0>\n"
+                "# For details, see the LICENSE file\n"
+                "# Commercial use beyond a 30-day trial requires a separate license\n\n"
+            )
+            if not valid_code.startswith("# Copyright (c)"):
+                valid_code = license_header + valid_code
+        # -------------------------------------------
+
         # Note: In an actual workflow we may want to inject this into the file. For now we overwrite/create.
         if target_file.exists():
             target_file.write_text(valid_code, encoding="utf-8")

src/coreason_meta_engineering/mcp_server.py

@@ -9,6 +9,7 @@
 # Source Code: <https://github.com/CoReason-AI/coreason-meta-engineering>
 import re
 import typing
+from datetime import UTC
 
 from coreason_manifest.spec import CognitiveDeliberativeEnvelopeState
 from mcp.server.fastmcp import FastMCP
@@ -231,5 +232,83 @@ def verify_solver_diff(
     return receipt.model_dump()
 
 
+@mcp.tool()  # type: ignore[misc]
+def scaffold_manifest_yaml(
+    target_dir: str,
+    urn: str,
+    author_id: str,
+) -> str:
+    """Scaffolds a new manifest.yaml for a given URN, injecting default CLA properties based on the AST Guillotine."""
+    import os
+    from datetime import datetime
+
+    import hvac
+    import yaml
+    from coreason_manifest.spec.ontology import COREASON_GLOBAL_TENANT_CID
+
+    vault_url = os.environ.get("VAULT_ADDR", "http://127.0.0.1:8200")
+    vault_token = os.environ.get("VAULT_TOKEN", "dev-only-token")
+
+    # Always try to identify the local environment that physically forged the asset
+    developer_tenant_cid = "UNKNOWN_LOCAL_TENANT"
+    private_cid = None
+    try:
+        client = hvac.Client(url=vault_url, token=vault_token)
+        response = client.secrets.kv.v2.read_secret_version(path="coreason/identity", raise_on_deleted_version=False)
+        if response and "data" in response and "data" in response["data"]:
+            ident = response["data"]["data"]
+            private_cid = ident.get("tenant_cid")
+            if private_cid:
+                developer_tenant_cid = private_cid
+    except Exception as e:
+        import logging
+        logging.getLogger(__name__).warning(f"Failed to fetch developer identity from Vault: {e}")
+
+    # AST Guillotine checks: defaults to CoReason Global for IP Ownership
+    cla_status = "UNSIGNED"
+    cla_assignee = ""
+    tenant_cid = COREASON_GLOBAL_TENANT_CID
+
+    if os.environ.get("AST_GUILLOTINE_ACTIVE") == "True":
+        cla_status = "AUTO_ASSIGNED_PPL3"
+        cla_assignee = "urn:tenant:coreason:global:authority"
+    else:
+        # Commercial Exception Active - Tenant keeps the IP they forged
+        if private_cid:
+            tenant_cid = private_cid
+            cla_assignee = private_cid
+
+    manifest_data = {
+        "urn": urn,
+        "tenant_cid": tenant_cid,
+        "default_clearance_tiers": [200],
+        "default_minimum_rigidity_tier": 255,
+        "epistemic_status": "DRAFT",
+        "provenance": {
+            "author_id": author_id,
+            "created_at": datetime.now(UTC).isoformat(),
+            "oracle_validator": None,
+            "certification": "pending",
+            "prior_event_hash": None,
+            "cla_status": cla_status,
+            "cla_assignee": cla_assignee,
+            "cla_version": "v1.0",
+            "developer_tenant_cid": developer_tenant_cid,
+            "cla_attestation_signature": "null",
+        },
+        "validation": {"test_coverage_pct": 0.0, "latency_ms": 0, "cryptographic_hash": "null"},
+    }
+
+    import pathlib
+
+    target = pathlib.Path(target_dir) / "manifest.yaml"
+    target.parent.mkdir(parents=True, exist_ok=True)
+
+    with open(target, "w", encoding="utf-8") as f:
+        yaml.dump(manifest_data, f, default_flow_style=False, sort_keys=False)
+
+    return f"Scaffolded manifest.yaml at {target}"
+
+
 def main() -> None:  # pragma: no cover
     mcp.run()

src/coreason_meta_engineering/pvv.py

@@ -106,13 +106,13 @@ def _compare_schema(module: Any, target_schema: dict[str, Any] | list[dict[str,
 
     if isinstance(target_schema, dict) and target_schema:
         target_properties = target_schema.get("properties", {})
-        
+
         # Try to find a model that has all required properties
         missing_keys = []
         for model in found_models:
             model_schema = model.model_json_schema()
             model_properties = model_schema.get("properties", {})
-            
+
             missing = [key for key in target_properties if key not in model_properties]
             if not missing:
                 return  # Found a valid model