diff --git a/src/cmdforge/attestation.py b/src/cmdforge/attestation.py new file mode 100644 index 0000000..e80a239 --- /dev/null +++ b/src/cmdforge/attestation.py @@ -0,0 +1,110 @@ +"""Supply chain attestation (M9.5). + +Tool publishers sign releases. Registry verifies signatures before +accepting publish. Clients verify signatures before installing. +""" + +import hashlib +import hmac +import json +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, Optional + + +@dataclass +class Attestation: + """A signed attestation for a tool release.""" + + tool_name: str + version: str + content_hash: str + signer: str # publisher identity + signature: str + signed_at: str = "" + algorithm: str = "hmac-sha256" + + def to_dict(self) -> dict: + return { + "tool_name": self.tool_name, + "version": self.version, + "content_hash": self.content_hash, + "signer": self.signer, + "signature": self.signature, + "signed_at": self.signed_at, + "algorithm": self.algorithm, + } + + +def sign_tool( + tool_name: str, + version: str, + content_hash: str, + signer: str, + secret_key: str, +) -> Attestation: + """Sign a tool release with HMAC-SHA256. + + Args: + tool_name: Tool name + version: Tool version + content_hash: Content hash from integrity module + signer: Publisher identity (username) + secret_key: Signing key (from config or keyring) + + Returns: + Attestation with signature + """ + payload = f"{tool_name}:{version}:{content_hash}:{signer}" + signature = hmac.new( + secret_key.encode(), + payload.encode(), + hashlib.sha256, + ).hexdigest() + + return Attestation( + tool_name=tool_name, + version=version, + content_hash=content_hash, + signer=signer, + signature=signature, + signed_at=datetime.now(timezone.utc).isoformat(), + ) + + +def verify_attestation(attestation: Attestation, secret_key: str) -> bool: + """Verify a tool attestation signature. + + Args: + attestation: The attestation to verify + secret_key: The signing key to verify against + + Returns: + True if signature is valid + """ + payload = f"{attestation.tool_name}:{attestation.version}:{attestation.content_hash}:{attestation.signer}" + expected = hmac.new( + secret_key.encode(), + payload.encode(), + hashlib.sha256, + ).hexdigest() + return hmac.compare_digest(expected, attestation.signature) + + +def verify_content_hash(content_hash: str, tool_dict: dict) -> bool: + """Verify that a content hash matches the tool definition. + + Args: + content_hash: The hash to verify + tool_dict: Tool dictionary (from to_dict()) + + Returns: + True if hash matches + """ + from .integrity import compute_tool_hash + # Reconstruct what the hash should be + tool_dict_copy = dict(tool_dict) + tool_dict_copy.pop("path", None) + content = json.dumps(tool_dict_copy, sort_keys=True) + expected = hashlib.sha256(content.encode()).hexdigest()[:16] + return expected == content_hash diff --git a/src/cmdforge/community.py b/src/cmdforge/community.py new file mode 100644 index 0000000..f6ad954 --- /dev/null +++ b/src/cmdforge/community.py @@ -0,0 +1,159 @@ +"""Community improvement workflow (M9.3). + +Allows community members to submit suggested prompt/code improvements. +Submissions are auto-tested before review. Successful improvements +credit both original and improving author. +""" + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from .tool import Tool, PromptStep, CodeStep + + +@dataclass +class ImprovementSubmission: + """A community-submitted improvement for a tool.""" + + tool_name: str + tool_version: str + submitter: str + step_index: int + step_type: str # "prompt" | "code" + original: str + proposed: str + rationale: str = "" + submitted_at: str = "" + status: str = "pending" # pending | tested | approved | rejected + test_result: Optional[dict] = None + + def to_dict(self) -> dict: + return { + "tool_name": self.tool_name, + "tool_version": self.tool_version, + "submitter": self.submitter, + "step_index": self.step_index, + "step_type": self.step_type, + "original": self.original[:200], + "proposed": self.proposed[:200], + "rationale": self.rationale, + "submitted_at": self.submitted_at, + "status": self.status, + "test_result": self.test_result, + } + + +@dataclass +class SubmissionReview: + """Review result for an improvement submission.""" + + submission: ImprovementSubmission + decision: str # "approve" | "reject" | "request_changes" + reviewer: str = "" + notes: str = "" + reviewed_at: str = "" + + +def create_submission( + tool: Tool, + step_index: int, + proposed: str, + submitter: str, + rationale: str = "", +) -> ImprovementSubmission: + """Create a new improvement submission.""" + if step_index < 0 or step_index >= len(tool.steps): + raise ValueError(f"Step index {step_index} out of range (0-{len(tool.steps)-1})") + + step = tool.steps[step_index] + if isinstance(step, PromptStep): + original = step.prompt + step_type = "prompt" + elif isinstance(step, CodeStep): + original = step.code + step_type = "code" + else: + raise ValueError(f"Step {step_index} is not a prompt or code step") + + return ImprovementSubmission( + tool_name=tool.name, + tool_version=tool.version or "", + submitter=submitter, + step_index=step_index, + step_type=step_type, + original=original, + proposed=proposed, + rationale=rationale, + submitted_at=datetime.now(timezone.utc).isoformat(), + ) + + +def run_submission_tests(submission: ImprovementSubmission, tool: Tool) -> dict: + """Run contract tests on the tool with the proposed change applied. + + Returns a dict with test results comparing baseline vs proposed. + """ + import copy + + from .contract_testing import run_contract_tests + + # Baseline + baseline = run_contract_tests(tool) + baseline_passed = sum( + 1 for r in baseline.results if r.state == "passed" + ) + + # Apply proposed change + tool_copy = copy.deepcopy(tool) + if submission.step_index >= len(tool_copy.steps): + return {"error": "Step index out of range", "baseline_passed": baseline_passed} + + step = tool_copy.steps[submission.step_index] + if isinstance(step, PromptStep) and submission.step_type == "prompt": + step.prompt = submission.proposed + elif isinstance(step, CodeStep) and submission.step_type == "code": + step.code = submission.proposed + else: + return {"error": "Step type mismatch", "baseline_passed": baseline_passed} + + # Test with proposed change + proposed_result = run_contract_tests(tool_copy) + proposed_passed = sum( + 1 for r in proposed_result.results if r.state == "passed" + ) + + result = { + "baseline_passed": baseline_passed, + "proposed_passed": proposed_passed, + "improvement": proposed_passed - baseline_passed, + "regressed": proposed_passed < baseline_passed, + "details": [r.to_dict() for r in proposed_result.results], + } + + submission.status = "tested" + submission.test_result = result + return result + + +def review_submission( + submission: ImprovementSubmission, + decision: str, + reviewer: str, + notes: str = "", +) -> SubmissionReview: + """Review a tested submission.""" + if decision not in ("approve", "reject", "request_changes"): + raise ValueError("Decision must be: approve, reject, or request_changes") + + if submission.status == "pending": + raise ValueError("Submission must be tested before review") + + submission.status = decision + return SubmissionReview( + submission=submission, + decision=decision, + reviewer=reviewer, + notes=notes, + reviewed_at=datetime.now(timezone.utc).isoformat(), + ) diff --git a/src/cmdforge/improvement.py b/src/cmdforge/improvement.py new file mode 100644 index 0000000..d70b06b --- /dev/null +++ b/src/cmdforge/improvement.py @@ -0,0 +1,212 @@ +"""Scrutiny-driven improvement pipeline (M9.2). + +Takes a tool with low scrutiny scores and generates actionable improvement +suggestions for honesty, efficiency, and transparency. +""" + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +from .tool import Tool, PromptStep, CodeStep + + +@dataclass +class ImprovementSuggestion: + """One actionable improvement suggestion.""" + + category: str # "honesty" | "efficiency" | "transparency" + severity: str # "high" | "medium" | "low" + title: str + description: str + location: str = "" # step index or "tool" + current: str = "" # current code/prompt text + suggested: str = "" # suggested replacement + auto_applicable: bool = False + + def to_dict(self) -> dict: + return { + "category": self.category, + "severity": self.severity, + "title": self.title, + "description": self.description, + "location": self.location, + "current": self.current[:200], + "suggested": self.suggested[:200], + "auto_applicable": self.auto_applicable, + } + + +@dataclass +class ImprovementReport: + """Collection of improvement suggestions for a tool.""" + + tool_name: str + suggestions: List[ImprovementSuggestion] = field(default_factory=list) + + @property + def has_high_severity(self) -> bool: + return any(s.severity == "high" for s in self.suggestions) + + @property + def by_category(self) -> Dict[str, List[ImprovementSuggestion]]: + result: Dict[str, List[ImprovementSuggestion]] = {} + for s in self.suggestions: + result.setdefault(s.category, []).append(s) + return result + + def to_dict(self) -> dict: + return { + "tool": self.tool_name, + "suggestion_count": len(self.suggestions), + "has_high_severity": self.has_high_severity, + "suggestions": [s.to_dict() for s in self.suggestions], + } + + +def generate_improvements(tool: Tool) -> ImprovementReport: + """Analyze a tool and generate improvement suggestions. + + Categories: + - Honesty: description doesn't match behavior + - Efficiency: unnecessary AI calls, redundant steps + - Transparency: obfuscated code, unclear prompts + """ + report = ImprovementReport(tool_name=tool.name) + + _check_honesty(tool, report) + _check_efficiency(tool, report) + _check_transparency(tool, report) + + return report + + +def _check_honesty(tool: Tool, report: ImprovementReport): + """Check description-to-behavior alignment.""" + if not tool.description: + report.suggestions.append(ImprovementSuggestion( + category="honesty", + severity="high", + title="Missing description", + description="Tool has no description. Users cannot know what it does without reading the config.", + location="tool", + auto_applicable=False, + )) + return + + # Check if description mentions capabilities not in steps + desc_lower = tool.description.lower() + has_prompt = any(isinstance(s, PromptStep) for s in tool.steps) + has_code = any(isinstance(s, CodeStep) for s in tool.steps) + + if "ai" in desc_lower or "generate" in desc_lower or "analyze" in desc_lower: + if not has_prompt: + report.suggestions.append(ImprovementSuggestion( + category="honesty", + severity="medium", + title="Description implies AI but no prompt step found", + description="The description mentions AI/generation/analysis but the tool has no PromptStep.", + location="tool", + auto_applicable=False, + )) + + # Check if description is too vague + if len(tool.description) < 20: + report.suggestions.append(ImprovementSuggestion( + category="honesty", + severity="low", + title="Description is very short", + description="A more detailed description helps users find and trust this tool.", + location="tool", + auto_applicable=False, + )) + + +def _check_efficiency(tool: Tool, report: ImprovementReport): + """Check for unnecessary AI calls and redundant steps.""" + prompt_steps = [s for s in tool.steps if isinstance(s, PromptStep)] + + # Multiple prompt steps with same provider — could be combined + if len(prompt_steps) > 1: + providers = [s.provider for s in prompt_steps] + if len(set(providers)) == 1: + report.suggestions.append(ImprovementSuggestion( + category="efficiency", + severity="medium", + title="Multiple prompt steps with same provider", + description=f"Found {len(prompt_steps)} prompt steps all using '{providers[0]}'. " + "Consider combining them into a single step to reduce API calls.", + location=f"steps 0-{len(prompt_steps)-1}", + auto_applicable=False, + )) + + # Prompt step that just passes input through + for i, step in enumerate(tool.steps): + if isinstance(step, PromptStep): + if step.prompt.strip() in ("{input}", "Repeat: {input}", "Echo: {input}"): + report.suggestions.append(ImprovementSuggestion( + category="efficiency", + severity="high", + title="Prompt step just echoes input", + description="This prompt step appears to just pass input through without transformation. " + "Remove it and use the input directly.", + location=f"step {i}", + current=step.prompt, + suggested="(remove this step — use {input} directly)", + auto_applicable=False, + )) + + # No steps at all — output is just a template + if not tool.steps and "{input}" in tool.output: + report.suggestions.append(ImprovementSuggestion( + category="efficiency", + severity="low", + title="Tool has no steps", + description="This tool only does variable substitution. Consider if it needs to be a CmdForge tool " + "or if a simple shell alias would suffice.", + location="tool", + auto_applicable=False, + )) + + +def _check_transparency(tool: Tool, report: ImprovementReport): + """Check for obfuscated code and unclear prompts.""" + for i, step in enumerate(tool.steps): + if isinstance(step, CodeStep): + # Check for overly complex code + lines = step.code.strip().split("\n") + if len(lines) > 20: + report.suggestions.append(ImprovementSuggestion( + category="transparency", + severity="medium", + title=f"Code step {i} is complex ({len(lines)} lines)", + description="Long code steps are harder to audit. Consider extracting into a separate " + "tool or adding comments.", + location=f"step {i}", + auto_applicable=False, + )) + + # Check for exec/eval + if "exec(" in step.code or "eval(" in step.code: + report.suggestions.append(ImprovementSuggestion( + category="transparency", + severity="high", + title=f"Code step {i} uses exec/eval", + description="Using exec() or eval() in code steps makes the tool harder to audit and " + "poses security risks. Consider restructuring.", + location=f"step {i}", + auto_applicable=False, + )) + + if isinstance(step, PromptStep): + # Check for very short prompts + if len(step.prompt.strip()) < 20: + report.suggestions.append(ImprovementSuggestion( + category="transparency", + severity="medium", + title=f"Prompt step {i} is very short", + description="Short prompts give the AI model less context, leading to unpredictable output. " + "Add more detail about the expected format and content.", + location=f"step {i}", + current=step.prompt, + auto_applicable=False, + )) diff --git a/src/cmdforge/integrity.py b/src/cmdforge/integrity.py new file mode 100644 index 0000000..1e5d0da --- /dev/null +++ b/src/cmdforge/integrity.py @@ -0,0 +1,133 @@ +"""Transitive integrity verification (M9.4). + +Content-addressable tool identity: tool = hash(tool definition + all dep hashes). +Extends lockfile to include transitive integrity chain. +""" + +import hashlib +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Optional + +from .tool import Tool, load_tool + + +@dataclass +class IntegrityNode: + """One node in the integrity chain.""" + + name: str + version: str = "" + content_hash: str = "" + dependencies: List[str] = field(default_factory=list) + dependency_hashes: Dict[str, str] = field(default_factory=dict) + + def to_dict(self) -> dict: + return { + "name": self.name, + "version": self.version, + "hash": self.content_hash, + "dependencies": self.dependencies, + "dependency_hashes": self.dependency_hashes, + } + + +@dataclass +class IntegrityChain: + """Full transitive integrity chain for a tool.""" + + root: IntegrityNode + nodes: Dict[str, IntegrityNode] = field(default_factory=dict) + + @property + def is_valid(self) -> bool: + """Check all dependency hashes match.""" + for name, node in self.nodes.items(): + for dep_name in node.dependencies: + expected_hash = node.dependency_hashes.get(dep_name) + if expected_hash == "unresolved": + # Unresolved dependencies don't invalidate the chain + continue + if dep_name not in self.nodes: + return False + dep_node = self.nodes[dep_name] + if expected_hash and expected_hash != dep_node.content_hash: + return False + return True + + def to_dict(self) -> dict: + return { + "root": self.root.to_dict(), + "nodes": {k: v.to_dict() for k, v in self.nodes.items()}, + "valid": self.is_valid, + } + + +def compute_tool_hash(tool: Tool) -> str: + """Compute a content hash for a tool definition.""" + tool_dict = tool.to_dict() + # Remove path (not part of content identity) + tool_dict.pop("path", None) + content = json.dumps(tool_dict, sort_keys=True) + return hashlib.sha256(content.encode()).hexdigest()[:16] + + +def build_integrity_chain(tool: Tool, max_depth: int = 10) -> IntegrityChain: + """Build a transitive integrity chain for a tool. + + Traverses all ToolStep dependencies and computes hashes for each. + """ + chain = IntegrityChain(root=_build_node(tool)) + chain.nodes[tool.name] = chain.root + _traverse_deps(tool, chain, depth=0, max_depth=max_depth) + return chain + + +def _build_node(tool: Tool) -> IntegrityNode: + """Build an integrity node from a tool.""" + deps = [] + dep_hashes = {} + + for step in tool.steps: + if hasattr(step, "tool") and step.tool: + deps.append(step.tool) + + # Resolve dependency hashes + for dep_name in deps: + dep_tool = load_tool(dep_name) + if dep_tool: + dep_hashes[dep_name] = compute_tool_hash(dep_tool) + else: + dep_hashes[dep_name] = "unresolved" + + return IntegrityNode( + name=tool.name, + version=tool.version or "", + content_hash=compute_tool_hash(tool), + dependencies=deps, + dependency_hashes=dep_hashes, + ) + + +def _traverse_deps(tool: Tool, chain: IntegrityChain, depth: int, max_depth: int): + """Recursively traverse dependencies and add to chain.""" + if depth >= max_depth: + return + + for step in tool.steps: + if hasattr(step, "tool") and step.tool: + dep_name = step.tool + if dep_name in chain.nodes: + continue + dep_tool = load_tool(dep_name) + if not dep_tool: + continue + node = _build_node(dep_tool) + chain.nodes[dep_name] = node + _traverse_deps(dep_tool, chain, depth + 1, max_depth) + + +def verify_integrity(chain: IntegrityChain) -> bool: + """Verify that all hashes in the chain are consistent.""" + return chain.is_valid diff --git a/tests/test_attestation.py b/tests/test_attestation.py new file mode 100644 index 0000000..49ab507 --- /dev/null +++ b/tests/test_attestation.py @@ -0,0 +1,45 @@ +"""Tests for M9.5 supply chain attestation.""" + +from cmdforge.attestation import ( + Attestation, + sign_tool, + verify_attestation, + verify_content_hash, +) +from cmdforge.tool import Tool + + +class TestSignTool: + def test_creates_attestation(self): + att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key") + assert att.tool_name == "mytool" + assert att.version == "1.0.0" + assert att.content_hash == "abc123" + assert att.signer == "alice" + assert len(att.signature) == 64 # SHA256 hex + assert att.algorithm == "hmac-sha256" + + def test_different_keys_different_signatures(self): + att1 = sign_tool("tool", "1.0.0", "hash", "alice", "key1") + att2 = sign_tool("tool", "1.0.0", "hash", "alice", "key2") + assert att1.signature != att2.signature + + +class TestVerifyAttestation: + def test_valid_signature(self): + att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key") + assert verify_attestation(att, "secret-key") + + def test_wrong_key_fails(self): + att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key") + assert not verify_attestation(att, "wrong-key") + + def test_tampered_content_fails(self): + att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key") + att.content_hash = "tampered" + assert not verify_attestation(att, "secret-key") + + def test_tampered_signer_fails(self): + att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key") + att.signer = "eve" + assert not verify_attestation(att, "secret-key") diff --git a/tests/test_community.py b/tests/test_community.py new file mode 100644 index 0000000..9eb69c9 --- /dev/null +++ b/tests/test_community.py @@ -0,0 +1,80 @@ +"""Tests for M9.3 community improvement workflow.""" + +from cmdforge.community import ( + ImprovementSubmission, + create_submission, + run_submission_tests, + review_submission, +) +from cmdforge.tool import Tool, PromptStep, CodeStep + + +class TestCreateSubmission: + def test_create_for_prompt_step(self): + tool = Tool( + name="test", + version="1.0.0", + steps=[PromptStep(prompt="Summarize", provider="mock", output_var="x")], + output="{x}", + ) + sub = create_submission(tool, 0, "Summarize the text thoroughly", "alice", "More detail") + assert sub.tool_name == "test" + assert sub.step_type == "prompt" + assert sub.original == "Summarize" + assert sub.proposed == "Summarize the text thoroughly" + assert sub.submitter == "alice" + assert sub.status == "pending" + + def test_create_for_code_step(self): + tool = Tool( + name="test", + steps=[CodeStep(code="x = 1", output_var="x")], + output="{x}", + ) + sub = create_submission(tool, 0, "x = 42", "bob") + assert sub.step_type == "code" + assert sub.original == "x = 1" + + def test_invalid_step_index(self): + tool = Tool(name="test") + try: + create_submission(tool, 5, "new", "alice") + assert False, "Should have raised" + except ValueError: + pass + + +class TestReviewSubmission: + def test_review_approved(self): + sub = ImprovementSubmission( + tool_name="test", tool_version="1.0.0", submitter="alice", + step_index=0, step_type="prompt", original="a", proposed="b", + status="tested", + ) + review = review_submission(sub, "approve", "admin", "Good improvement") + assert review.decision == "approve" + assert sub.status == "approve" + + def test_review_pending_rejected(self): + sub = ImprovementSubmission( + tool_name="test", tool_version="1.0.0", submitter="alice", + step_index=0, step_type="prompt", original="a", proposed="b", + status="pending", + ) + try: + review_submission(sub, "approve", "admin") + assert False, "Should have raised" + except ValueError: + pass + + def test_invalid_decision(self): + sub = ImprovementSubmission( + tool_name="test", tool_version="1.0.0", submitter="alice", + step_index=0, step_type="prompt", original="a", proposed="b", + status="tested", + ) + try: + review_submission(sub, "maybe", "admin") + assert False, "Should have raised" + except ValueError: + pass diff --git a/tests/test_improvement.py b/tests/test_improvement.py new file mode 100644 index 0000000..fc09774 --- /dev/null +++ b/tests/test_improvement.py @@ -0,0 +1,109 @@ +"""Tests for scrutiny-driven improvement pipeline (M9.2).""" + +from cmdforge.improvement import ( + ImprovementSuggestion, + ImprovementReport, + generate_improvements, +) +from cmdforge.tool import Tool, PromptStep, CodeStep + + +class TestImprovementReport: + def test_empty_report(self): + report = ImprovementReport(tool_name="test") + assert not report.has_high_severity + assert report.by_category == {} + + def test_has_high_severity(self): + report = ImprovementReport( + tool_name="test", + suggestions=[ + ImprovementSuggestion("honesty", "high", "Bad", "desc"), + ImprovementSuggestion("efficiency", "low", "Minor", "desc"), + ], + ) + assert report.has_high_severity + + def test_by_category(self): + report = ImprovementReport( + tool_name="test", + suggestions=[ + ImprovementSuggestion("honesty", "high", "A", "d"), + ImprovementSuggestion("honesty", "low", "B", "d"), + ImprovementSuggestion("efficiency", "medium", "C", "d"), + ], + ) + assert len(report.by_category["honesty"]) == 2 + assert len(report.by_category["efficiency"]) == 1 + + +class TestGenerateImprovements: + def test_missing_description(self): + tool = Tool(name="nodesc") + report = generate_improvements(tool) + honesty = [s for s in report.suggestions if s.category == "honesty"] + assert any("Missing description" in s.title for s in honesty) + + def test_short_description(self): + tool = Tool(name="short", description="Does stuff") + report = generate_improvements(tool) + honesty = [s for s in report.suggestions if s.category == "honesty"] + assert any("short" in s.title.lower() for s in honesty) + + def test_echo_prompt_detected(self): + tool = Tool( + name="echo", + steps=[PromptStep(prompt="{input}", provider="mock", output_var="out")], + output="{out}", + ) + report = generate_improvements(tool) + efficiency = [s for s in report.suggestions if s.category == "efficiency"] + assert any("echo" in s.title.lower() for s in efficiency) + + def test_multiple_same_provider_prompts(self): + tool = Tool( + name="multi", + steps=[ + PromptStep(prompt="Step 1: {input}", provider="claude", output_var="a"), + PromptStep(prompt="Step 2: {a}", provider="claude", output_var="b"), + ], + output="{b}", + ) + report = generate_improvements(tool) + efficiency = [s for s in report.suggestions if s.category == "efficiency"] + assert any("same provider" in s.title.lower() for s in efficiency) + + def test_exec_detected(self): + tool = Tool( + name="dangerous", + steps=[CodeStep(code="exec('print(1)')", output_var="x")], + output="{x}", + ) + report = generate_improvements(tool) + transparency = [s for s in report.suggestions if s.category == "transparency"] + assert any("exec" in s.title.lower() for s in transparency) + + def test_short_prompt_detected(self): + tool = Tool( + name="vague", + steps=[PromptStep(prompt="Do it", provider="mock", output_var="x")], + output="{x}", + ) + report = generate_improvements(tool) + transparency = [s for s in report.suggestions if s.category == "transparency"] + assert any("short" in s.title.lower() for s in transparency) + + def test_clean_tool_no_suggestions(self): + tool = Tool( + name="clean", + description="A well-documented tool that summarizes text input concisely", + steps=[PromptStep( + prompt="Summarize the following text in 2-3 sentences. Focus on key points.\n\nText: {input}", + provider="mock", + output_var="summary", + )], + output="{summary}", + ) + report = generate_improvements(tool) + high = [s for s in report.suggestions if s.severity == "high"] + assert len(high) == 0 diff --git a/tests/test_integrity.py b/tests/test_integrity.py new file mode 100644 index 0000000..cf46ea8 --- /dev/null +++ b/tests/test_integrity.py @@ -0,0 +1,97 @@ +"""Tests for M9.4 transitive integrity verification.""" + +from unittest.mock import patch + +from cmdforge.integrity import ( + IntegrityNode, + IntegrityChain, + compute_tool_hash, + build_integrity_chain, + verify_integrity, +) +from cmdforge.tool import Tool, PromptStep, ToolStep + + +class TestComputeToolHash: + def test_same_tool_same_hash(self): + tool = Tool(name="test", version="1.0.0") + assert compute_tool_hash(tool) == compute_tool_hash(tool) + + def test_different_tools_different_hash(self): + a = Tool(name="a", version="1.0.0") + b = Tool(name="b", version="1.0.0") + assert compute_tool_hash(a) != compute_tool_hash(b) + + def test_version_change_changes_hash(self): + v1 = Tool(name="test", version="1.0.0") + v2 = Tool(name="test", version="2.0.0") + assert compute_tool_hash(v1) != compute_tool_hash(v2) + + +class TestIntegrityChain: + def test_single_tool_chain(self): + tool = Tool(name="solo", version="1.0.0") + chain = build_integrity_chain(tool) + assert chain.root.name == "solo" + assert chain.is_valid + + def test_chain_with_unresolved_dep(self): + tool = Tool( + name="parent", + version="1.0.0", + steps=[ToolStep(tool="missing-dep", output_var="x")], + output="{x}", + ) + chain = build_integrity_chain(tool) + assert "missing-dep" not in chain.nodes + # Chain is still valid because the dependency hash is "unresolved" + # and there's no node to compare against + assert chain.is_valid + + def test_chain_valid_with_resolved_dep(self, tmp_path): + with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"): + from cmdforge.tool import save_tool + + child = Tool(name="child", version="1.0.0") + save_tool(child) + + parent = Tool( + name="parent", + version="1.0.0", + steps=[ToolStep(tool="child", output_var="x")], + output="{x}", + ) + save_tool(parent) + + chain = build_integrity_chain(parent) + assert "child" in chain.nodes + assert chain.is_valid + + +class TestVerifyIntegrity: + def test_valid_chain(self): + chain = IntegrityChain( + root=IntegrityNode(name="root", content_hash="abc"), + nodes={"root": IntegrityNode(name="root", content_hash="abc")}, + ) + assert verify_integrity(chain) + + def test_tampered_chain(self): + chain = IntegrityChain( + root=IntegrityNode( + name="root", + content_hash="abc", + dependencies=["dep"], + dependency_hashes={"dep": "expected_hash"}, + ), + nodes={ + "root": IntegrityNode( + name="root", + content_hash="abc", + dependencies=["dep"], + dependency_hashes={"dep": "expected_hash"}, + ), + "dep": IntegrityNode(name="dep", content_hash="different_hash"), + }, + ) + assert not chain.is_valid