Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .dockerignore
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
.git/
# .git/
.venv/
__pycache__/
.pytest_cache/
Expand Down
7 changes: 2 additions & 5 deletions src/coreason_meta_engineering/forge_orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@

from coreason_meta_engineering.pvv import execute_pvv_pipeline
from coreason_meta_engineering.utils.logger import logger
from coreason_meta_engineering.utils.mcp_registry_scaffolder import generate_server_json
# Legacy generate_server_json import removed

__all__ = ["DynamicForgeOrchestrator", "dispatch_agent_generation", "orchestrate_generation"]

Expand Down Expand Up @@ -189,10 +189,7 @@ async def scaffold_ast(
target_file.parent.mkdir(parents=True, exist_ok=True)
target_file.write_text(valid_code, encoding="utf-8")

# --- Dual-Publish: Generate server.json ---
server_json_content = generate_server_json(geometric_schema)
server_json_path = target_file.parent / "server.json"
server_json_path.write_text(server_json_content, encoding="utf-8")
# Legacy Dual-Publish (server.json) removed in favor of OCI-native distribution.

return valid_code

Expand Down
107 changes: 75 additions & 32 deletions src/coreason_meta_engineering/utils/mcp_registry_scaffolder.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,39 +8,82 @@
#
# Source Code: <https://github.com/CoReason-AI/coreason-meta-engineering>

import json
from typing import Any
import subprocess

Check notice

Code scanning / Bandit

Consider possible security implications associated with the subprocess module. Note

Consider possible security implications associated with the subprocess module.
from pathlib import Path
from typing import List

from .logger import logger

def generate_server_json(geometric_schema: dict[str, Any] | list[dict[str, Any]]) -> str:

def generate_push_command(
component_path: Path,
oci_uri: str,
insecure: bool = False,
) -> List[str]:
"""
AGENT INSTRUCTION: Generates the deterministic 'wash push' command for OCI artifact publishing.
This ensures standard WASI 0.2 components are pushed to standard registries with proper annotations.

CAUSAL AFFORDANCE: Enables the orchestrator to publish artifacts to OCI registries.

EPISTEMIC BOUNDS: Operates on Path objects and URI strings. Enforces --insecure flag based on policy.
Injects mandatory OCI annotations for traceability.

MCP ROUTING TRIGGERS: OCI, WASI, wash, push, registry, artifact
"""
command = [
"wash",
"push",
oci_uri,
component_path.as_posix(),
"--annotation",
"org.opencontainers.image.source=https://github.com/coreason-ai/coreason",
]
if insecure:
command.append("--insecure")
return command


def publish_mcp_artifact(
component_path: Path,
oci_uri: str,
insecure: bool = False,
) -> str:
"""
Generates an MCP Registry server.json file based on the geometric schema.
This guarantees type isomorphism and prevents LLM hallucination.
AGENT INSTRUCTION: Executes the 'wash push' CLI command to publish a WASI component.
This is the kinetic layer of artifact distribution.

CAUSAL AFFORDANCE: Synchronously pushes a local .wasm component to a remote OCI registry.

EPISTEMIC BOUNDS: Requires the 'wash' CLI to be present in the execution environment.
Enforces strict .wasm extension check to guarantee WASI 0.2 component integrity.
Wraps subprocess.run with strict error handling.

MCP ROUTING TRIGGERS: OCI, WASI, wash, push, registry, artifact, kinetic
"""
if isinstance(geometric_schema, list):
geometric_schema = geometric_schema[0] if geometric_schema else {}

urn = geometric_schema.get("urn", "urn:coreason:actionspace:unknown:v1")

# Extract components of URN assuming urn:coreason:actionspace:{type}:{name}:{version}
parts = urn.split(":")
capability_name = parts[4] if len(parts) >= 6 else "unknown_capability"

server_json = {
"name": capability_name,
"description": geometric_schema.get("description", f"CoReason capabilities for {capability_name}"),
"version": "1.0.0",
"mcp_version": "1.0",
"author": "CoReason, Inc.",
"license": geometric_schema.get("economics", {}).get("license_class", "PROSPERITY_3.0_COMMERCIAL"),
"entrypoint": "mcp-server",
"repository": f"https://github.com/CoReason-AI/{capability_name}",
"registry_metadata": {
"urn": urn,
"epistemic_status": geometric_schema.get("epistemic_status", "DRAFT"),
"billing_tier": geometric_schema.get("economics", {}).get("billing_tier", "TIER_1_STANDARD"),
"validation": geometric_schema.get("validation", {}),
},
}

return json.dumps(server_json, indent=2)
if component_path.suffix != ".wasm":
raise ValueError(
f"Strict WASI 0.2 enforcement: Capability artifacts must be compiled .wasm binaries. Got: {component_path.suffix}"
)

if not component_path.exists():
raise FileNotFoundError(f"Component not found at {component_path}")

command = generate_push_command(component_path, oci_uri, insecure)
logger.info(f"Publishing artifact to {oci_uri}...")

try:
result = subprocess.run(
command,

Check notice

Code scanning / Bandit

subprocess call - check for execution of untrusted input. Note

subprocess call - check for execution of untrusted input.
Comment on lines +76 to +77
capture_output=True,
text=True,
check=True,
)
logger.info(f"Successfully pushed {oci_uri}")
return result.stdout
except subprocess.CalledProcessError as e:
logger.error(f"Failed to push {oci_uri}: {e.stderr}")
raise RuntimeError(f"wash push failed: {e.stderr}") from e
except FileNotFoundError:
logger.error("The 'wash' CLI was not found in the environment.")
raise RuntimeError("The 'wash' CLI is required for OCI publishing but was not found.")
96 changes: 56 additions & 40 deletions tests/utils/test_mcp_registry_scaffolder.py
Original file line number Diff line number Diff line change
@@ -1,40 +1,56 @@
import json
from coreason_meta_engineering.utils.mcp_registry_scaffolder import generate_server_json

def test_generate_server_json_dict():
schema = {
"urn": "urn:coreason:actionspace:solver:test_solver:v1",
"description": "Test description",
"epistemic_status": "PUBLISHED",
"economics": {
"license_class": "OPEN_SOURCE",
"billing_tier": "TIER_0_FREE"
},
"validation": {}
}
result = generate_server_json(schema)
data = json.loads(result)
assert data["name"] == "test_solver"
assert data["registry_metadata"]["epistemic_status"] == "PUBLISHED"

def test_generate_server_json_list():
schema = [{
"urn": "urn:coreason:actionspace:solver:test_list_solver:v1"
}]
result = generate_server_json(schema)
data = json.loads(result)
assert data["name"] == "test_list_solver"

def test_generate_server_json_empty_list():
schema = []
result = generate_server_json(schema)
data = json.loads(result)
assert data["name"] == "unknown_capability"

def test_generate_server_json_invalid_urn():
schema = {
"urn": "invalid_urn"
}
result = generate_server_json(schema)
data = json.loads(result)
assert data["name"] == "unknown_capability"
# Copyright (c) 2026 CoReason, Inc
#
# This software is proprietary and dual-licensed
# Licensed under the Prosperity Public License 3.0 (the "License")
# A copy of the license is available at <https://prosperitylicense.com/versions/3.0.0>
# For details, see the LICENSE file
# Commercial use beyond a 30-day trial requires a separate license
#
# Source Code: <https://github.com/CoReason-AI/coreason-meta-engineering>

from pathlib import Path
import pytest
from coreason_meta_engineering.utils.mcp_registry_scaffolder import generate_push_command, publish_mcp_artifact


def test_generate_push_command_basic() -> None:
"""
Asserts that the wash push command is generated correctly with mandatory annotations.
"""
component_path = Path("build/test_solver.wasm")
oci_uri = "ghcr.io/coreason-ai/test-solver:v1"

command = generate_push_command(component_path, oci_uri)

assert command == [
"wash",
"push",
"ghcr.io/coreason-ai/test-solver:v1",
"build/test_solver.wasm",
"--annotation",
"org.opencontainers.image.source=https://github.com/coreason-ai/coreason",
]


def test_generate_push_command_insecure() -> None:
"""
Asserts that the --insecure flag is appended correctly for local/untrusted registries.
"""
component_path = Path("build/test_solver.wasm")
oci_uri = "localhost:5000/test-solver:v1"

command = generate_push_command(component_path, oci_uri, insecure=True)

assert "--insecure" in command
assert "localhost:5000/test-solver:v1" in command


def test_publish_mcp_artifact_invalid_extension() -> None:
"""
Asserts that the publisher enforces strict WASI 0.2 component validation (.wasm suffix).
"""
component_path = Path("src/solver.py")
oci_uri = "ghcr.io/coreason-ai/test-solver:v1"

with pytest.raises(ValueError, match="Strict WASI 0.2 enforcement"):
publish_mcp_artifact(component_path, oci_uri)
Loading