diff --git a/src/howlcreate/cli/main.py b/src/howlcreate/cli/main.py index 7e56eae..9db9b6f 100644 --- a/src/howlcreate/cli/main.py +++ b/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 diff --git a/src/howlcreate/engine/candidate_ingestion.py b/src/howlcreate/engine/candidate_ingestion.py new file mode 100644 index 0000000..b55f652 --- /dev/null +++ b/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", + }, + } diff --git a/tests/test_candidate_ingestion.py b/tests/test_candidate_ingestion.py new file mode 100644 index 0000000..6953c59 --- /dev/null +++ b/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