src/pqc_reasoning_ledger/step.py
5.1 KB · 144 lines · python Raw
1 """ReasoningStep - one hash-chained unit of thought."""
2
3 from __future__ import annotations
4
5 import hashlib
6 import json
7 import uuid
8 from dataclasses import asdict, dataclass, field
9 from datetime import datetime, timezone
10 from enum import Enum
11 from typing import Any
12
13
14 class StepKind(str, Enum):
15 """Types of reasoning steps - the symbolic vocabulary."""
16
17 THOUGHT = "thought" # free-form reasoning statement
18 OBSERVATION = "observation" # observation about input or retrieved data
19 HYPOTHESIS = "hypothesis" # a tentative conclusion
20 DEDUCTION = "deduction" # logical deduction from prior steps
21 RETRIEVAL = "retrieval" # fetching external knowledge
22 TOOL_CALL = "tool-call" # calling an external tool / function
23 TOOL_RESULT = "tool-result"
24 SELF_CRITIQUE = "self-critique" # model critiquing its own prior step
25 REFINEMENT = "refinement" # updated answer after critique
26 DECISION = "decision" # final decision / answer
27 META = "meta" # metadata about the run
28
29
30 @dataclass(frozen=True)
31 class StepReference:
32 """A reference from one step to an earlier step
33 (e.g. 'deduction references observation X')."""
34
35 step_id: str
36 relationship: str # "depends-on" | "refutes" | "refines" | "cites"
37
38 def to_dict(self) -> dict[str, Any]:
39 return asdict(self)
40
41
42 @dataclass
43 class ReasoningStep:
44 """One step in a chain-of-thought reasoning trace.
45
46 Every step is hashed with:
47 SHA3-256( previous_hash || canonical_bytes(step_payload) )
48 Steps chain via previous_step_hash, so any tampering of an intermediate
49 step invalidates every step after it.
50 """
51
52 step_id: str
53 step_number: int # 1-based position within the trace
54 kind: StepKind
55 content: str # the actual reasoning text
56 timestamp: str
57 content_hash: str = "" # SHA3-256 of content
58 step_hash: str = "" # chain hash: SHA3-256(prev_hash || canonical_bytes)
59 previous_step_hash: str = "0" * 64
60 references: list[StepReference] = field(default_factory=list)
61 confidence: float = 1.0 # 0..1 model's reported confidence in this step
62 metadata: dict[str, Any] = field(default_factory=dict)
63
64 @staticmethod
65 def hash_content(content: str) -> str:
66 return hashlib.sha3_256(content.encode("utf-8")).hexdigest()
67
68 def canonical_bytes(self) -> bytes:
69 """Deterministic payload for hashing - excludes chain hash."""
70 payload = {
71 "step_id": self.step_id,
72 "step_number": self.step_number,
73 "kind": self.kind.value,
74 "content_hash": self.content_hash,
75 "timestamp": self.timestamp,
76 "previous_step_hash": self.previous_step_hash,
77 "references": [r.to_dict() for r in self.references],
78 "confidence": self.confidence,
79 "metadata": self.metadata,
80 }
81 return json.dumps(
82 payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False
83 ).encode("utf-8")
84
85 def compute_step_hash(self) -> str:
86 """SHA3-256 over (previous_step_hash || canonical_bytes)."""
87 prev = (
88 bytes.fromhex(self.previous_step_hash)
89 if self.previous_step_hash
90 else b"\x00" * 32
91 )
92 return hashlib.sha3_256(prev + self.canonical_bytes()).hexdigest()
93
94 @classmethod
95 def create(
96 cls,
97 kind: StepKind,
98 content: str,
99 step_number: int,
100 previous_step_hash: str = "0" * 64,
101 references: list[StepReference] | None = None,
102 confidence: float = 1.0,
103 metadata: dict[str, Any] | None = None,
104 ) -> ReasoningStep:
105 step_id = f"urn:pqc-step:{uuid.uuid4().hex}"
106 now = datetime.now(timezone.utc).isoformat()
107 content_hash = cls.hash_content(content)
108 step = cls(
109 step_id=step_id,
110 step_number=step_number,
111 kind=kind,
112 content=content,
113 timestamp=now,
114 content_hash=content_hash,
115 step_hash="",
116 previous_step_hash=previous_step_hash,
117 references=list(references or []),
118 confidence=confidence,
119 metadata=dict(metadata or {}),
120 )
121 step.step_hash = step.compute_step_hash()
122 return step
123
124 def to_dict(self) -> dict[str, Any]:
125 d = asdict(self)
126 d["kind"] = self.kind.value
127 return d
128
129 @classmethod
130 def from_dict(cls, data: dict[str, Any]) -> ReasoningStep:
131 return cls(
132 step_id=data["step_id"],
133 step_number=int(data["step_number"]),
134 kind=StepKind(data["kind"]),
135 content=data["content"],
136 timestamp=data["timestamp"],
137 content_hash=data.get("content_hash", ""),
138 step_hash=data.get("step_hash", ""),
139 previous_step_hash=data.get("previous_step_hash", "0" * 64),
140 references=[StepReference(**r) for r in data.get("references", [])],
141 confidence=float(data.get("confidence", 1.0)),
142 metadata=dict(data.get("metadata", {})),
143 )
144