coulomb/markitect-quarkdown
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Quarkdown render adapter boundary

Browse Source
This commit is contained in:
Bernd Worsch
2026-05-15 14:46:32 +02:00
parent 0063fd8fcc
commit 63258161e6
11 changed files with 1139 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
src/markitect_quarkdown/__init__.py Normal file
View File

@@ -0,0 +1,29 @@
"""Quarkdown render/export adapter for Markitect."""
from markitect_quarkdown.adapter import (
MARKITECT_QUARKDOWN_VERSION_POLICY,
QUARKDOWN_PROFILE_MATRIX,
MarkitectQuarkdownAdapter,
QuarkdownExecutionPlan,
QuarkdownProcessResult,
QuarkdownProfileMapping,
build_quarkdown_execution_plan,
export_quarkdown_source,
quarkdown_adapter_descriptor,
quarkdown_profile_for,
validate_quarkdown_artifact,
)
__all__ = [
"MARKITECT_QUARKDOWN_VERSION_POLICY",
"QUARKDOWN_PROFILE_MATRIX",
"MarkitectQuarkdownAdapter",
"QuarkdownExecutionPlan",
"QuarkdownProcessResult",
"QuarkdownProfileMapping",
"build_quarkdown_execution_plan",
"export_quarkdown_source",
"quarkdown_adapter_descriptor",
"quarkdown_profile_for",
"validate_quarkdown_artifact",
]

631
src/markitect_quarkdown/adapter.py Normal file
View File

@@ -0,0 +1,631 @@
"""Concrete Quarkdown render/export adapter boundary."""
from __future__ import annotations
from dataclasses import asdict, dataclass, field
import hashlib
import json
from pathlib import Path
import shutil
import subprocess
import tempfile
from typing import Any, Callable
from markitect_tool.diagnostics import Diagnostic, SourceLocation, has_error
from markitect_tool.extension import OptionalDependency
from markitect_tool.render import (
RenderArtifact,
RenderExportAdapterDescriptor,
RenderExportRequest,
RenderExportResult,
RenderProvenance,
)
MARKITECT_QUARKDOWN_VERSION_POLICY = "quarkdown-2.x"
QUARKDOWN_ADAPTER_ID = "render.quarkdown"
@dataclass(frozen=True)
class QuarkdownProfileMapping:
"""Mapping from a Markitect render profile to Quarkdown invocation intent."""
markitect_profile: str
quarkdown_document_type: str
output_format: str
artifact_media_type: str
artifact_extension: str
pdf_export: bool = False
notes: str | None = None
def to_dict(self) -> dict[str, Any]:
return _drop_empty(asdict(self))
QUARKDOWN_PROFILE_MATRIX: dict[str, QuarkdownProfileMapping] = {
"plain": QuarkdownProfileMapping(
markitect_profile="plain",
quarkdown_document_type="plain",
output_format="txt",
artifact_media_type="text/plain",
artifact_extension=".txt",
),
"docs": QuarkdownProfileMapping(
markitect_profile="docs",
quarkdown_document_type="docs",
output_format="html",
artifact_media_type="text/html",
artifact_extension=".html",
),
"slides": QuarkdownProfileMapping(
markitect_profile="slides",
quarkdown_document_type="slides",
output_format="html",
artifact_media_type="text/html",
artifact_extension=".html",
),
"paged": QuarkdownProfileMapping(
markitect_profile="paged",
quarkdown_document_type="paged",
output_format="pdf",
artifact_media_type="application/pdf",
artifact_extension=".pdf",
pdf_export=True,
),
"static-site": QuarkdownProfileMapping(
markitect_profile="static-site",
quarkdown_document_type="docs",
output_format="html",
artifact_media_type="text/html",
artifact_extension=".html",
notes="Quarkdown docs output plus static/public asset handling.",
),
"pdf": QuarkdownProfileMapping(
markitect_profile="pdf",
quarkdown_document_type="paged",
output_format="pdf",
artifact_media_type="application/pdf",
artifact_extension=".pdf",
pdf_export=True,
),
}
@dataclass(frozen=True)
class QuarkdownExecutionPlan:
"""Concrete, inspectable Quarkdown CLI execution plan."""
command: list[str]
source_path: str
output_dir: str
expected_artifact_path: str
profile: str
quarkdown_document_type: str
output_format: str
permissions: list[str] = field(default_factory=list)
environment: dict[str, str] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
return _drop_empty(asdict(self))
@dataclass(frozen=True)
class QuarkdownProcessResult:
"""Result returned by a Quarkdown command runner."""
returncode: int
stdout: str = ""
stderr: str = ""
metadata: dict[str, Any] = field(default_factory=dict)
@property
def successful(self) -> bool:
return self.returncode == 0
def to_dict(self) -> dict[str, Any]:
return _drop_empty(asdict(self))
QuarkdownCommandResolver = Callable[[str], str | None]
QuarkdownCommandRunner = Callable[[QuarkdownExecutionPlan], QuarkdownProcessResult]
class MarkitectQuarkdownAdapter:
"""Render/export adapter that invokes Quarkdown through a controlled boundary."""
def __init__(
self,
*,
command: str = "quarkdown",
command_resolver: QuarkdownCommandResolver | None = None,
runner: QuarkdownCommandRunner | None = None,
) -> None:
self.command = command
self.command_resolver = command_resolver or shutil.which
self.runner = runner or _subprocess_runner
self.descriptor = quarkdown_adapter_descriptor()
def render(self, request: RenderExportRequest) -> RenderExportResult:
adapter_info = _adapter_info()
diagnostics = _validate_request(request)
if diagnostics:
return RenderExportResult(
adapter=adapter_info,
operation=request.operation,
profile=request.profile,
diagnostics=diagnostics,
)
mapping = quarkdown_profile_for(request.profile)
exported_source = export_quarkdown_source(request, mapping=mapping)
source_artifact = RenderArtifact.from_content(
exported_source,
role="renderer-source",
media_type="text/markdown",
metadata={
"quarkdown_document_type": mapping.quarkdown_document_type,
"profile": request.profile,
},
)
provenance = [
RenderProvenance(
operation=request.operation,
adapter_id=QUARKDOWN_ADAPTER_ID,
profile=request.profile,
source_path=request.source_path,
source_digest=_digest_text(request.source),
artifact_id=source_artifact.artifact_id,
metadata={"version_policy": MARKITECT_QUARKDOWN_VERSION_POLICY},
)
]
if request.operation == "inspect-profile":
return RenderExportResult(
adapter=adapter_info,
operation=request.operation,
profile=request.profile,
provenance=provenance,
metadata={
"profile": mapping.to_dict(),
"option_schema": _option_schema(),
"version_policy": MARKITECT_QUARKDOWN_VERSION_POLICY,
"external_renderer_invoked": False,
},
)
if request.operation == "export-source":
return RenderExportResult(
adapter=adapter_info,
operation=request.operation,
profile=request.profile,
artifacts=[source_artifact],
exported_source=exported_source,
provenance=provenance,
metadata={
"external_renderer_invoked": False,
"quarkdown_document_type": mapping.quarkdown_document_type,
},
)
command_path = self.command_resolver(str(request.options.get("command", self.command)))
if not command_path:
return RenderExportResult(
adapter=adapter_info,
operation=request.operation,
profile=request.profile,
artifacts=[source_artifact],
exported_source=exported_source,
provenance=provenance,
diagnostics=[
_diagnostic(
"render.quarkdown.runtime_missing",
"Quarkdown CLI is not available on PATH.",
details={
"command": request.options.get("command", self.command),
"runtime_dependencies": ["Quarkdown CLI", "Java 17+", "Node.js/npm/Puppeteer for PDF"],
},
)
],
metadata={"external_renderer_invoked": False},
)
with tempfile.TemporaryDirectory(prefix="markitect-quarkdown-") as temp_dir:
workspace = Path(request.options.get("workspace") or temp_dir)
source_path = _write_renderer_source(workspace, exported_source)
output_dir = Path(request.options.get("output_dir") or workspace / "quarkdown-output")
output_dir.mkdir(parents=True, exist_ok=True)
plan = build_quarkdown_execution_plan(
request,
source_path=source_path,
output_dir=output_dir,
command=command_path,
mapping=mapping,
)
if request.options.get("dry_run"):
plan_artifact = RenderArtifact.from_content(
json.dumps(plan.to_dict(), indent=2, sort_keys=True),
role="execution-plan",
media_type="application/json",
metadata={"external_renderer_invoked": False},
)
return RenderExportResult(
adapter=adapter_info,
operation=request.operation,
profile=request.profile,
artifacts=[source_artifact, plan_artifact],
exported_source=exported_source,
provenance=provenance,
metadata={"plan": plan.to_dict(), "external_renderer_invoked": False},
)
process_result = self.runner(plan)
if not process_result.successful:
return RenderExportResult(
adapter=adapter_info,
operation=request.operation,
profile=request.profile,
artifacts=[source_artifact],
exported_source=exported_source,
provenance=provenance,
diagnostics=[
_diagnostic(
"render.quarkdown.execution_failed",
"Quarkdown CLI execution failed.",
details=process_result.to_dict() | {"plan": plan.to_dict()},
)
],
metadata={"external_renderer_invoked": True},
)
artifact, artifact_diagnostics = validate_quarkdown_artifact(plan)
if artifact:
provenance = [
RenderProvenance(
operation=request.operation,
adapter_id=QUARKDOWN_ADAPTER_ID,
profile=request.profile,
source_path=request.source_path,
source_digest=_digest_text(request.source),
artifact_id=artifact.artifact_id,
metadata={
"version_policy": MARKITECT_QUARKDOWN_VERSION_POLICY,
"plan": plan.to_dict(),
},
)
]
return RenderExportResult(
adapter=adapter_info,
operation=request.operation,
profile=request.profile,
artifacts=[source_artifact, *([artifact] if artifact else [])],
exported_source=exported_source,
diagnostics=artifact_diagnostics,
provenance=provenance,
metadata={
"external_renderer_invoked": True,
"process": process_result.to_dict(),
"plan": plan.to_dict(),
},
)
def quarkdown_adapter_descriptor() -> RenderExportAdapterDescriptor:
"""Return the Quarkdown render/export adapter descriptor."""
return RenderExportAdapterDescriptor(
id=QUARKDOWN_ADAPTER_ID,
version="1",
name="Quarkdown Render Adapter",
summary="Concrete Quarkdown CLI render/export adapter boundary for Markitect.",
operations=["inspect-profile", "export-source", "render-artifact"],
input_contracts=["Markdown", "RenderReferenceManifest", "DocumentFunctionEvaluationResult"],
output_profiles=list(QUARKDOWN_PROFILE_MATRIX),
artifact_media_types=sorted({mapping.artifact_media_type for mapping in QUARKDOWN_PROFILE_MATRIX.values()}),
factory=MarkitectQuarkdownAdapter,
optional_dependencies=[
OptionalDependency(
name="quarkdown-cli",
package="Quarkdown 2.x",
required=False,
purpose="Compile Quarkdown source to HTML/PDF/slides/docs artifacts.",
),
OptionalDependency(
name="java",
package="Java 17+",
required=False,
purpose="Quarkdown runtime.",
),
OptionalDependency(
name="node",
package="Node.js/npm/Puppeteer",
required=False,
purpose="PDF export through Quarkdown's HTML rendering pipeline.",
),
],
option_schema=_option_schema(),
safety={
"filesystem_read": True,
"filesystem_write": True,
"external_process": True,
"network": "permission-controlled",
"native_renderer_dependency": True,
"assisted_generation": False,
"publication_side_effect": False,
},
quality_profile={
"renderer": "quarkdown-2.x",
"pdf_export": "requires-node-npm-puppeteer",
"runtime_dependencies_optional": True,
"tests_skip_without_runtime": True,
},
metadata={
"version_policy": MARKITECT_QUARKDOWN_VERSION_POLICY,
"profile_matrix": [mapping.to_dict() for mapping in QUARKDOWN_PROFILE_MATRIX.values()],
"permission_flags": ["project-read", "global-read", "network", "native-content", "all"],
"default_output_dir": "quarkdown-output",
"open_reuse_integration": "integration/quarkdown.integration.yaml",
},
)
def quarkdown_profile_for(profile: str) -> QuarkdownProfileMapping:
"""Return Quarkdown mapping for a Markitect render profile."""
try:
return QUARKDOWN_PROFILE_MATRIX[profile]
except KeyError as exc:
raise ValueError(f"Unsupported Markitect render profile `{profile}`") from exc
def export_quarkdown_source(
request: RenderExportRequest,
*,
mapping: QuarkdownProfileMapping | None = None,
) -> str:
"""Return Quarkdown source exported from a Markitect render request."""
mapping = mapping or quarkdown_profile_for(request.profile)
header = [
"<!-- generated-by: markitect-quarkdown -->",
f"<!-- markitect-profile: {request.profile} -->",
f"<!-- quarkdown-document-type: {mapping.quarkdown_document_type} -->",
]
manifest_id = _manifest_id(request.render_manifest)
if manifest_id:
header.append(f"<!-- render-reference-manifest: {manifest_id} -->")
return "\n".join(header) + "\n\n" + request.source
def build_quarkdown_execution_plan(
request: RenderExportRequest,
*,
source_path: str | Path,
output_dir: str | Path,
command: str = "quarkdown",
mapping: QuarkdownProfileMapping | None = None,
) -> QuarkdownExecutionPlan:
"""Build an inspectable Quarkdown CLI command plan."""
mapping = mapping or quarkdown_profile_for(request.profile)
source_file = Path(source_path)
target_dir = Path(output_dir)
artifact_stem = str(request.options.get("artifact_stem") or source_file.stem or "document")
expected_artifact = target_dir / f"{artifact_stem}{mapping.artifact_extension}"
permissions = _permissions_for(request)
command_args = [
command,
"compile",
str(source_file),
"--output",
str(target_dir),
"--format",
mapping.output_format,
"--type",
mapping.quarkdown_document_type,
]
for permission in permissions:
command_args.extend(["--allow", permission])
if request.options.get("deny_network", True):
command_args.extend(["--deny", "network"])
return QuarkdownExecutionPlan(
command=command_args,
source_path=str(source_file),
output_dir=str(target_dir),
expected_artifact_path=str(expected_artifact),
profile=request.profile,
quarkdown_document_type=mapping.quarkdown_document_type,
output_format=mapping.output_format,
permissions=permissions,
environment={
"requires_java": "17+",
"requires_node_for_pdf": str(mapping.pdf_export).lower(),
},
metadata={
"pdf_export": mapping.pdf_export,
"version_policy": MARKITECT_QUARKDOWN_VERSION_POLICY,
},
)
def validate_quarkdown_artifact(
plan: QuarkdownExecutionPlan,
) -> tuple[RenderArtifact | None, list[Diagnostic]]:
"""Validate the expected Quarkdown artifact and return Markitect metadata."""
path = Path(plan.expected_artifact_path)
mapping = quarkdown_profile_for(plan.profile)
diagnostics: list[Diagnostic] = []
if not path.exists() or not path.is_file():
diagnostics.append(
_diagnostic(
"render.quarkdown.artifact_missing",
"Expected Quarkdown artifact was not produced.",
details={"path": str(path), "plan": plan.to_dict()},
)
)
return None, diagnostics
if path.suffix.lower() != mapping.artifact_extension:
diagnostics.append(
_diagnostic(
"render.quarkdown.artifact_extension",
"Quarkdown artifact extension does not match the requested profile.",
details={
"path": str(path),
"expected_extension": mapping.artifact_extension,
"actual_extension": path.suffix.lower(),
},
)
)
if path.stat().st_size <= 0:
diagnostics.append(
_diagnostic(
"render.quarkdown.artifact_empty",
"Quarkdown artifact is empty.",
details={"path": str(path)},
)
)
artifact = RenderArtifact(
artifact_id=f"artifact:{_file_digest(path).removeprefix('sha256:')[:16]}",
role="rendered-artifact",
media_type=mapping.artifact_media_type,
path=str(path),
digest=_file_digest(path),
metadata={
"quarkdown_document_type": plan.quarkdown_document_type,
"output_format": plan.output_format,
"version_policy": MARKITECT_QUARKDOWN_VERSION_POLICY,
},
)
return artifact, diagnostics
def _option_schema() -> dict[str, Any]:
return {
"type": "object",
"properties": {
"command": {"type": "string", "default": "quarkdown"},
"workspace": {"type": "string"},
"output_dir": {"type": "string", "default": "quarkdown-output"},
"artifact_stem": {"type": "string", "default": "document"},
"permissions": {
"type": "array",
"items": {"type": "string"},
"default": ["project-read"],
},
"deny_network": {"type": "boolean", "default": True},
"dry_run": {"type": "boolean", "default": False},
},
"additionalProperties": True,
}
def _permissions_for(request: RenderExportRequest) -> list[str]:
permissions = request.options.get("permissions", ["project-read"])
if isinstance(permissions, str):
permissions = [permissions]
return [str(permission) for permission in permissions]
def _validate_request(request: RenderExportRequest) -> list[Diagnostic]:
diagnostics: list[Diagnostic] = []
if request.profile not in QUARKDOWN_PROFILE_MATRIX:
diagnostics.append(
_diagnostic(
"render.quarkdown.profile_unsupported",
f"Quarkdown adapter does not support profile `{request.profile}`.",
details={"profile": request.profile},
)
)
if request.operation not in {"inspect-profile", "export-source", "render-artifact"}:
diagnostics.append(
_diagnostic(
"render.quarkdown.operation_unsupported",
f"Quarkdown adapter does not support operation `{request.operation}`.",
details={"operation": request.operation},
)
)
if not request.source and request.operation != "inspect-profile":
diagnostics.append(
_diagnostic("render.quarkdown.source_missing", "Quarkdown render source cannot be empty.")
)
if request.operation == "render-artifact":
blocked = []
policy = request.policy or {}
for capability in ("filesystem_write", "external_process", "native_renderer_dependency"):
if policy.get(capability) is False:
blocked.append(capability)
if blocked:
diagnostics.append(
_diagnostic(
"render.quarkdown.capability_blocked",
"Quarkdown render execution requires blocked capabilities.",
details={"capabilities": blocked},
)
)
return diagnostics
def _write_renderer_source(workspace: Path, source: str) -> Path:
workspace.mkdir(parents=True, exist_ok=True)
source_path = workspace / "markitect-source.qd"
source_path.write_text(source, encoding="utf-8")
return source_path
def _subprocess_runner(plan: QuarkdownExecutionPlan) -> QuarkdownProcessResult:
completed = subprocess.run(
plan.command,
cwd=Path(plan.output_dir).parent,
text=True,
capture_output=True,
check=False,
)
return QuarkdownProcessResult(
returncode=completed.returncode,
stdout=completed.stdout,
stderr=completed.stderr,
)
def _adapter_info() -> dict[str, str]:
return {"id": QUARKDOWN_ADAPTER_ID, "version": "1", "name": "Quarkdown Render Adapter"}
def _manifest_id(manifest: Any) -> str | None:
if manifest is None:
return None
if hasattr(manifest, "to_dict"):
data = manifest.to_dict()
elif isinstance(manifest, dict):
data = manifest
else:
return None
value = data.get("manifest_id")
return str(value) if value else None
def _diagnostic(code: str, message: str, *, details: dict[str, Any] | None = None) -> Diagnostic:
return Diagnostic(
severity="error",
code=code,
message=message,
source=SourceLocation(path="<quarkdown>"),
details=details or {},
)
def _digest_text(value: str) -> str:
return "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest()
def _file_digest(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return "sha256:" + digest.hexdigest()
def _drop_empty(data: dict[str, Any]) -> dict[str, Any]:
return {key: value for key, value in data.items() if value not in (None, [], {}, "")}