decomposer: generate deliverable files for Inspect the single_agent template and target repository conventions to identify the required project structure, configuration, interfaces, and implementation patterns for the application landing zone agent.; Define the application landing zone agent's behavior, input and output contracts, document-generation workflow, and architecture-diagram requirements using the extracted template conventions.; Validate the implemented application_landing_zone_agent by running repository tests and checking the TSD document generation, draw.io diagram generation, agent packaging, configuration, and committed implementation against the defined contracts.; Commit and push the validated application_landing_zone_agent implementation to the target repository.

src/application_landing_zone_agent/__init__.py

@@ -0,0 +1,5 @@
+"""Application landing zone agent public API."""
+from .agent import LandingZoneAgent, GenerationResult, ValidationError
+
+__all__ = ["LandingZoneAgent", "GenerationResult", "ValidationError"]
+__version__ = "0.1.0"

src/application_landing_zone_agent/agent.py

@@ -0,0 +1,75 @@
+import html
+import json
+from pathlib import Path
+from typing import Any
+from .model import Component, Requirements, GenerationResult, ValidationReport
+
+class ValidationError(ValueError):
+    """Raised when requirements cannot satisfy the input contract."""
+
+class LandingZoneAgent:
+    def __init__(self, config: dict[str, Any] | None = None):
+        self.config = config or {}
+
+    def normalize(self, data: dict[str, Any]) -> Requirements:
+        report = self.validate(data)
+        if report.errors:
+            raise ValidationError("; ".join(report.errors))
+        components = tuple(Component(str(c["name"]), str(c.get("type", "component")), str(c.get("technology", "Not specified")), str(c.get("description", ""))) for c in data.get("components", []))
+        def strings(key: str) -> tuple[str, ...]:
+            return tuple(str(x) for x in data.get(key, []))
+        integrations = tuple(str(x.get("name", "")) if isinstance(x, dict) else str(x) for x in data.get("integrations", []))
+        return Requirements(str(data["application_name"]).strip(), str(data["business_context"]).strip(), strings("environments"), components, integrations, strings("constraints"), strings("non_functional_requirements"), strings("assumptions"), dict(data))
+
+    def validate(self, data: Any) -> ValidationReport:
+        errors: list[str] = []
+        if not isinstance(data, dict): return ValidationReport(False, ["Input must be a JSON object"])
+        for key in ("application_name", "business_context"):
+            if not isinstance(data.get(key), str) or not data[key].strip(): errors.append(f"'{key}' is required and must be a non-empty string")
+        for key in ("environments", "components", "integrations", "constraints", "non_functional_requirements", "assumptions"):
+            if key in data and not isinstance(data[key], list): errors.append(f"'{key}' must be an array")
+        for i, component in enumerate(data.get("components", [])):
+            if not isinstance(component, dict) or not isinstance(component.get("name"), str) or not component["name"].strip(): errors.append(f"components[{i}].name must be a non-empty string")
+        for i, item in enumerate(data.get("integrations", [])):
+            if isinstance(item, dict) and (not isinstance(item.get("name"), str) or not item["name"].strip()): errors.append(f"integrations[{i}].name must be a non-empty string")
+            elif not isinstance(item, (str, int, float)) or not str(item).strip(): errors.append(f"integrations[{i}] must be a string or named object")
+        warnings = []
+        if not data.get("components"): warnings.append("No application components supplied; diagram contains only the boundary")
+        return ValidationReport(not errors, errors, warnings)
+
+    def generate(self, data: dict[str, Any]) -> GenerationResult:
+        req = self.normalize(data)
+        report = self.validate(data)
+        tsd = self._tsd(req)
+        diagram = self._diagram(req)
+        if not all(h in tsd for h in ("# Technical Solution Design", "## Component inventory", "## Operational considerations")):
+            raise RuntimeError("TSD renderer omitted a required heading")
+        return GenerationResult(tsd, diagram, report)
+
+    def _bullets(self, values: tuple[str, ...]) -> str:
+        return "\n".join(f"- {v}" for v in values) if values else "- None provided"
+
+    def _tsd(self, r: Requirements) -> str:
+        components = "\n".join(f"| {c.name} | {c.type} | {c.technology} | {c.description or 'Not specified'} |" for c in r.components) or "| None provided | - | - | - |"
+        envs = ", ".join(r.environments) if r.environments else "Not specified"
+        return f"""# Technical Solution Design: {r.application_name}\n\n## Scope and business context\n{r.business_context}\n\n## Environment strategy\n{envs}\n\n## Component inventory\n| Name | Type | Technology | Description |\n|---|---|---|---|\n{components}\n\n## Integrations\n{self._bullets(r.integrations)}\n\n## Constraints\n{self._bullets(r.constraints)}\n\n## Non-functional requirements\n{self._bullets(r.non_functional_requirements)}\n\n## Security and data considerations\n- Apply least-privilege identity to each workload.\n- Keep secrets in a managed secret store; no credentials are embedded in this design.\n- Confirm data classification, retention, network boundaries, and ingress/egress policy during review.\n\n## Operational considerations\n- Define health checks, structured logs, metrics, alert thresholds, backup/restore, and rollback ownership before release.\n- Confirm availability, recovery objectives, capacity, and cost guardrails against the stated requirements.\n\n## Assumptions and decisions needed\n{self._bullets(r.assumptions)}\n- Decisions needed: hosting region, identity integration, data classification, and deployment ownership.\n"""
+
+    def _diagram(self, r: Requirements) -> str:
+        cells = ['<mxCell id="0"/>', '<mxCell id="1" parent="0"/>']
+        def cell_id(prefix: str, i: int) -> str: return f"{prefix}-{i}"
+        cells.append('<mxCell id="boundary" value="Application landing zone" style="swimlane;horizontal=0;rounded=1;" vertex="1" parent="1"><mxGeometry x="40" y="40" width="900" height="650" as="geometry"/></mxCell>')
+        for i, c in enumerate(r.components):
+            x, y = 90 + (i % 3) * 270, 130 + (i // 3) * 150
+            label = html.escape(f"{c.name}\\n{c.type}\\n{c.technology}")
+            cells.append(f'<mxCell id="{cell_id("component", i)}" value="{label}" style="rounded=1;whiteSpace=wrap;html=1;fillColor=#dae8fc;" vertex="1" parent="boundary"><mxGeometry x="{x}" y="{y}" width="210" height="80" as="geometry"/></mxCell>')
+        for i, name in enumerate(r.integrations):
+            label = html.escape(name)
+            cells.append(f'<mxCell id="{cell_id("integration", i)}" value="{label}" style="shape=cloud;whiteSpace=wrap;html=1;fillColor=#fff2cc;" vertex="1" parent="1"><mxGeometry x="1020" y="{120 + i * 130}" width="190" height="80" as="geometry"/></mxCell>')
+            if r.components: cells.append(f'<mxCell id="edge-integration-{i}" value="integration" style="edgeStyle=orthogonalEdgeStyle;rounded=0;" edge="1" parent="1" source="component-0" target="{cell_id("integration", i)}"><mxGeometry relative="1" as="geometry"/></mxCell>')
+        for i in range(max(0, len(r.components) - 1)):
+            cells.append(f'<mxCell id="edge-component-{i}" value="" style="edgeStyle=orthogonalEdgeStyle;rounded=0;" edge="1" parent="boundary" source="component-{i}" target="component-{i+1}"><mxGeometry relative="1" as="geometry"/></mxCell>')
+        body = "".join(cells)
+        return f'<?xml version="1.0" encoding="UTF-8"?><mxfile host="app.diagrams.net" version="24.7.17"><diagram id="landing-zone" name="Architecture"><mxGraphModel dx="1200" dy="800" grid="1" gridSize="10" page="1" pageWidth="1600" pageHeight="1000"><root>{body}</root></mxGraphModel></diagram></mxfile>'
+
+def load_config(path: str | None) -> dict[str, Any]:
+    return json.loads(Path(path).read_text()) if path else {}

src/application_landing_zone_agent/cli.py

@@ -0,0 +1,23 @@
+import argparse, json, sys
+from pathlib import Path
+from .agent import LandingZoneAgent, ValidationError, load_config
+
+def main(argv=None) -> int:
+    p = argparse.ArgumentParser(description="Generate a TSD and draw.io architecture diagram")
+    p.add_argument("--input", required=True, help="requirements JSON")
+    p.add_argument("--output-dir", required=True)
+    p.add_argument("--config", default="config/default.json")
+    args = p.parse_args(argv)
+    try:
+        data = json.loads(Path(args.input).read_text())
+        result = LandingZoneAgent(load_config(args.config)).generate(data)
+        out = Path(args.output_dir); out.mkdir(parents=True, exist_ok=True)
+        result_files = {"tsd.md": result.tsd_markdown, "architecture.drawio": result.drawio_xml, "validation.json": json.dumps(result.validation.__dict__, indent=2) + "\n"}
+        for name, content in result_files.items(): (out / name).write_text(content)
+        return 0
+    except ValidationError as exc:
+        print(f"validation error: {exc}", file=sys.stderr); return 2
+    except (OSError, json.JSONDecodeError, RuntimeError) as exc:
+        print(f"generation error: {exc}", file=sys.stderr); return 1
+
+if __name__ == "__main__": sys.exit(main())

src/application_landing_zone_agent/model.py

@@ -0,0 +1,33 @@
+from dataclasses import dataclass, field
+from typing import Any
+
+@dataclass(frozen=True)
+class ValidationReport:
+    valid: bool
+    errors: list[str] = field(default_factory=list)
+    warnings: list[str] = field(default_factory=list)
+
+@dataclass(frozen=True)
+class GenerationResult:
+    tsd_markdown: str
+    drawio_xml: str
+    validation: ValidationReport
+
+@dataclass(frozen=True)
+class Component:
+    name: str
+    type: str = "component"
+    technology: str = "Not specified"
+    description: str = ""
+
+@dataclass(frozen=True)
+class Requirements:
+    application_name: str
+    business_context: str
+    environments: tuple[str, ...] = ()
+    components: tuple[Component, ...] = ()
+    integrations: tuple[str, ...] = ()
+    constraints: tuple[str, ...] = ()
+    non_functional_requirements: tuple[str, ...] = ()
+    assumptions: tuple[str, ...] = ()
+    raw: dict[str, Any] = field(default_factory=dict)