feat(ingestion): add candidate ingestion and deliberate sandbox development

src/howlcreate/cli/main.py

@@ -191,6 +191,38 @@ def cmd_list(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_develop(args: argparse.Namespace) -> int:
+    """Ingest a candidate handoff and assessment, producing a deliberate development plan."""
+    from howlcreate.engine.candidate_ingestion import develop_candidate, IngestionError
+
+    cand_path = Path(args.candidate_file)
+    assess_path = Path(args.assessment)
+
+    if not cand_path.is_file():
+        print(f"Error: Candidate file '{cand_path}' not found", file=sys.stderr)
+        return 1
+    if not assess_path.is_file():
+        print(f"Error: Assessment file '{assess_path}' not found", file=sys.stderr)
+        return 1
+
+    cand_data = json.loads(cand_path.read_text(encoding="utf-8"))
+    assess_data = json.loads(assess_path.read_text(encoding="utf-8"))
+
+    try:
+        dev_res = develop_candidate(cand_data, assess_data)
+    except IngestionError as e:
+        print(f"Ingestion rejected: {e}", file=sys.stderr)
+        return 2
+
+    out_text = json.dumps(dev_res, indent=2)
+    if args.output:
+        Path(args.output).write_text(out_text, encoding="utf-8")
+        print(f"[Saved] Development plan written to: {args.output}")
+    else:
+        print(out_text)
+    return 0
+
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         prog="howlcreate",
@@ -248,6 +280,13 @@ def build_parser() -> argparse.ArgumentParser:
     list_parser = subparsers.add_parser("list", help="List all stored runs")
     list_parser.set_defaults(func=cmd_list)
 
+    # develop
+    develop_parser = subparsers.add_parser("develop", help="Ingest a HowlFrame-promoted candidate for deliberate sandbox development")
+    develop_parser.add_argument("candidate_file", type=str, help="Path to howl.candidate/v1 JSON file")
+    develop_parser.add_argument("--assessment", "-a", type=str, required=True, help="Path to howl.assessment/v1 JSON file")
+    develop_parser.add_argument("--output", "-o", type=str, help="Output destination file")
+    develop_parser.set_defaults(func=cmd_develop)
+
     return parser
 
 

src/howlcreate/engine/candidate_ingestion.py

@@ -0,0 +1,180 @@
+"""Candidate ingestion and deliberate sandbox development for HowlDream integration."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any, Dict
+
+from howlcreate.models.idea import (
+    ConceptStatus,
+    EpistemicStatus,
+    Idea,
+    LineageGraph,
+)
+
+
+class IngestionError(Exception):
+    """Raised when candidate ingestion or authority validation fails."""
+    pass
+
+
+def develop_candidate(
+    candidate: Dict[str, Any],
+    assessment: Dict[str, Any],
+) -> Dict[str, Any]:
+    """Ingest a HowlFrame-promoted candidate and produce a deliberate sandbox development plan.
+
+    HowlCreate treats this as speculative exploration, not ground truth.
+    It develops proposals in a sandbox; it does NOT deploy or mutate production.
+    """
+    # 1. Authority validation: fail closed on any execution attempt
+    auth = candidate.get("authority", {})
+    if auth.get("executable") is True or auth.get("type") != "ADVISORY":
+        raise IngestionError(
+            "Authority escalation prohibited: speculative candidate cannot claim execution authority"
+        )
+
+    assess_auth = assessment.get("authority", {})
+    if assess_auth.get("executable") is True or assess_auth.get("type") != "ADVISORY":
+        raise IngestionError(
+            "Authority escalation prohibited: assessment cannot claim execution authority"
+        )
+
+    # 2. Gate validation: must be ACCEPT_FOR_DEVELOPMENT
+    disposition = assessment.get("disposition")
+    if disposition != "ACCEPT_FOR_DEVELOPMENT":
+        raise IngestionError(
+            f"Candidate cannot be developed: disposition is {disposition!r}, "
+            "must be 'ACCEPT_FOR_DEVELOPMENT'"
+        )
+
+    candidate_id = candidate.get("candidate_id", "unknown_candidate")
+    objective = candidate.get("objective", "")
+    text = candidate.get("text", "")
+    assumptions = candidate.get("assumptions", [])
+    constraints = candidate.get("verified_constraints", [])
+    parent_req = candidate.get("parent_request_id", "")
+
+    # 3. Create Idea model with strict epistemic tagging
+    clean_id = candidate_id.replace("/", "-")
+    idea_id = f"create-{clean_id}"
+    title = f"Sandbox Prototype for {clean_id}"
+
+    idea = Idea(
+        id=idea_id,
+        title=title,
+        description=text,
+        problem_framing=objective,
+        core_mechanism=f"Exploratory mechanism based on candidate {candidate_id}",
+        operator_used="deliberate_sandbox_development",
+        parent_ids=[candidate_id],
+        origin="howldream",
+        epistemic_status=EpistemicStatus.IMAGINED_POSSIBILITY,
+        assumptions=list(assumptions),
+        constraints=list(constraints),
+        speculations=[
+            "Exploration hypothesis requires controlled sandbox validation before any production consideration."
+        ],
+        evidence_needs=[
+            f"Independent verification of assumptions: {', '.join(assumptions) if assumptions else 'None declared'}"
+        ],
+        unanswered_questions=[
+            "What are the precise latency and failure boundaries in staging?"
+        ],
+        status=ConceptStatus.CANDIDATE,
+    )
+
+    # 4. Record Lineage in LineageGraph
+    graph = LineageGraph()
+    parent_idea = Idea(
+        id=candidate_id,
+        title=f"DREAM Candidate ({candidate_id})",
+        description=text,
+        origin="howldream",
+        epistemic_status=EpistemicStatus.IMAGINED_POSSIBILITY,
+        operator_used="howldream_exploration",
+    )
+    graph.add_idea(parent_idea)
+    graph.add_idea(idea)
+    graph.add_edge(
+        candidate_id,
+        idea_id,
+        operator="deliberate_sandbox_development",
+        rationale="Candidate accepted by HowlFrame evaluation for deliberate sandbox development",
+    )
+
+    # 5. Formulate Deliberate Sandbox Plan (Design, Test Spec, Architecture Proposal)
+    prototype_design = {
+        "prototype_id": f"proto-{idea_id}",
+        "target_sandbox_environment": "isolated_local_testbed",
+        "implementation_steps": [
+            "1. Instantiate mock service harness simulating intermittent deployment failures.",
+            "2. Implement diagnostic hook matching candidate exploration proposal.",
+            "3. Execute fault-injection sweep without remote egress or infrastructure mutation.",
+        ],
+        "isolation_controls": [
+            "NO_PRODUCTION_DEPLOYMENT",
+            "NO_IMPLICIT_NETWORK_EGRESS",
+            "READ_ONLY_ACCESS_ONLY",
+        ],
+    }
+
+    test_specification = [
+        {
+            "test_id": "test_sandbox_diagnostic_activation",
+            "assertion": "Diagnostic captures failure metrics under simulated timeout",
+            "expected_outcome": "PASS",
+        },
+        {
+            "test_id": "test_sandbox_zero_side_effects",
+            "assertion": "Diagnostic produces zero non-reproducible state mutations",
+            "expected_outcome": "PASS",
+        },
+    ]
+
+    architecture_proposal = (
+        f"# Deliberate Sandbox Architecture Proposal for {idea_id}\n\n"
+        f"**Origin**: HowlDream Speculative Exploration ({candidate_id})\n"
+        f"**Evaluation**: HowlFrame ACCEPT_FOR_DEVELOPMENT ({assessment.get('assessment_id')})\n"
+        f"**Epistemic Status**: IMAGINED_POSSIBILITY (Exploratory Prototype Only)\n\n"
+        f"## Objective\n{objective}\n\n"
+        f"## Proposed Mechanism\n{text}\n\n"
+        f"## Verified Constraints\n"
+        + "".join(f"- {c}\n" for c in constraints)
+        + f"\n## Unresolved Assumptions\n"
+        + "".join(f"- {a}\n" for a in assumptions)
+        + "\n## Authority Boundary\n"
+        "This proposal represents deliberate design in sandbox isolation. "
+        "It carries NO EXECUTION OR DEPLOYMENT AUTHORITY in HowlPlane or HowlChangeOps."
+    )
+
+    now_str = datetime.now(timezone.utc).isoformat()
+    return {
+        "schema_version": "howl.development_result/v1",
+        "development_id": f"dev-{clean_id}",
+        "source_candidate_id": candidate_id,
+        "parent_request_id": parent_req,
+        "origin": "howldream",
+        "epistemic_status": EpistemicStatus.IMAGINED_POSSIBILITY.value,
+        "authority": {
+            "type": "ADVISORY",
+            "executable": False,
+        },
+        "execution_authority": "NONE",
+        "idea": idea.to_dict(),
+        "lineage": {
+            "parent_id": candidate_id,
+            "child_id": idea_id,
+            "ancestors": graph.get_ancestors(idea_id),
+        },
+        "sandbox_prototype_design": prototype_design,
+        "test_specification": test_specification,
+        "architecture_proposal": architecture_proposal,
+        "provenance": {
+            "candidate_id": candidate_id,
+            "assessment_id": assessment.get("assessment_id"),
+            "parent_request_id": parent_req,
+            "developed_at": now_str,
+            "system": "howlcreate",
+        },
+    }

tests/test_candidate_ingestion.py

@@ -0,0 +1,129 @@
+"""Tests for HowlCreate candidate ingestion and deliberate sandbox development."""
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+import pytest
+
+from howlcreate.engine.candidate_ingestion import develop_candidate, IngestionError
+
+
+def make_payloads():
+    candidate = {
+        "schema_version": "howl.candidate/v1",
+        "candidate_id": "run-20260911-001/candidates/0/1",
+        "source_run_id": "run-20260911-001",
+        "parent_request_id": "req-20260911-alpha",
+        "objective": "Explore alternative diagnostic strategies for deployment timeouts",
+        "text": "IDEA: eBPF socket tracing probes on staging ingress proxies to catch TCP RST packets during blue-green switchovers",
+        "condition": "dream",
+        "trust": "UNVERIFIED",
+        "status": "LOCALLY_VERIFIED",
+        "authority": {"type": "ADVISORY", "executable": False},
+        "claims": [
+            {"kind": "IDEA", "text": "eBPF socket tracing probes on staging ingress proxies"},
+            {"kind": "FACT", "text": "staging ingress runs Linux kernel 6.8+"},
+        ],
+        "assumptions": ["Kernel BTF support is enabled on staging proxies"],
+        "unresolved_issues": ["Overhead under 10k conn/sec unverified"],
+        "contradictions": [],
+        "verified_constraints": ["Kernel 6.8+ present"],
+    }
+    assessment = {
+        "schema_version": "howl.assessment/v1",
+        "assessment_id": "assess-run-20260911-001-c1",
+        "candidate_id": "run-20260911-001/candidates/0/1",
+        "disposition": "ACCEPT_FOR_DEVELOPMENT",
+        "confidence": "MEDIUM",
+        "reason": "sufficient evidence invariants passed for deliberate development",
+        "authority": {"type": "ADVISORY", "executable": False},
+    }
+    return candidate, assessment
+
+
+def test_develop_candidate_accepted():
+    cand, assess = make_payloads()
+    result = develop_candidate(cand, assess)
+
+    assert result["schema_version"] == "howl.development_result/v1"
+    assert result["source_candidate_id"] == cand["candidate_id"]
+    assert result["origin"] == "howldream"
+    assert result["epistemic_status"] == "IMAGINED_POSSIBILITY"
+    assert result["authority"]["executable"] is False
+    assert result["authority"]["type"] == "ADVISORY"
+    assert result["execution_authority"] == "NONE"
+
+    # Verify idea
+    idea = result["idea"]
+    assert idea["origin"] == "howldream"
+    assert idea["parent_ids"] == [cand["candidate_id"]]
+
+    # Verify lineage
+    lineage = result["lineage"]
+    assert lineage["parent_id"] == cand["candidate_id"]
+    assert cand["candidate_id"] in lineage["ancestors"]
+
+    # Verify sandbox plan
+    proto = result["sandbox_prototype_design"]
+    assert "isolated_local_testbed" in proto["target_sandbox_environment"]
+    assert "NO_PRODUCTION_DEPLOYMENT" in proto["isolation_controls"]
+
+
+def test_develop_candidate_rejected_fails_closed():
+    cand, assess = make_payloads()
+    assess["disposition"] = "REJECT"
+
+    with pytest.raises(IngestionError, match="must be 'ACCEPT_FOR_DEVELOPMENT'"):
+        develop_candidate(cand, assess)
+
+
+def test_develop_candidate_unresolved_fails_closed():
+    cand, assess = make_payloads()
+    assess["disposition"] = "UNRESOLVED"
+
+    with pytest.raises(IngestionError, match="must be 'ACCEPT_FOR_DEVELOPMENT'"):
+        develop_candidate(cand, assess)
+
+
+def test_develop_candidate_authority_escalation_fails_closed():
+    cand, assess = make_payloads()
+    cand["authority"]["executable"] = True
+
+    with pytest.raises(IngestionError, match="Authority escalation prohibited"):
+        develop_candidate(cand, assess)
+
+    cand, assess = make_payloads()
+    assess["authority"]["type"] = "EXECUTIVE"
+
+    with pytest.raises(IngestionError, match="Authority escalation prohibited"):
+        develop_candidate(cand, assess)
+
+
+def test_cli_develop(tmp_path: Path):
+    cand, assess = make_payloads()
+    cand_file = tmp_path / "candidate.json"
+    assess_file = tmp_path / "assessment.json"
+    out_file = tmp_path / "development_result.json"
+
+    cand_file.write_text(json.dumps(cand, indent=2))
+    assess_file.write_text(json.dumps(assess, indent=2))
+
+    cmd = [
+        sys.executable,
+        "-m",
+        "howlcreate.cli.main",
+        "develop",
+        str(cand_file),
+        "--assessment",
+        str(assess_file),
+        "--output",
+        str(out_file),
+    ]
+    res = subprocess.run(cmd, capture_output=True, text=True, check=True)
+    assert "[Saved]" in res.stdout
+    assert out_file.exists()
+
+    dev_data = json.loads(out_file.read_text())
+    assert dev_data["schema_version"] == "howl.development_result/v1"
+    assert dev_data["authority"]["executable"] is False