Append-only hash-chained audit log for one Mission. One chain per Mission. Initialized at CREATED state, closed at CLOSED state. Usage:: chain = AuditChain.initialize(mission_id="msn_...", agent_id="aid_...") chain = chain.append( layer=AuditLayer
| 153 | |
| 154 | |
| 155 | class AuditChain(BaseModel): |
| 156 | """ |
| 157 | Append-only hash-chained audit log for one Mission. |
| 158 | One chain per Mission. Initialized at CREATED state, closed at CLOSED state. |
| 159 | |
| 160 | Usage:: |
| 161 | |
| 162 | chain = AuditChain.initialize(mission_id="msn_...", agent_id="aid_...") |
| 163 | |
| 164 | chain = chain.append( |
| 165 | layer=AuditLayer.EXECUTION, |
| 166 | event="EXECUTION_tool_call", |
| 167 | data={"tool": "accounting_api", "result": "success"}, |
| 168 | token_id="tok_...", |
| 169 | ) |
| 170 | |
| 171 | ok = chain.verify_integrity() |
| 172 | |
| 173 | For database-backed persistence, use ``orgkernel.database`` or implement a custom |
| 174 | repository that stores AuditChainModel rows. |
| 175 | """ |
| 176 | chain_id: str = Field(default_factory=_new_chain_id) |
| 177 | mission_id: str |
| 178 | agent_id: str |
| 179 | initialized_at: datetime = Field(default_factory=_now_utc) |
| 180 | closed_at: datetime | None = None |
| 181 | entries: list[AuditEntry] = Field(default_factory=list) |
| 182 | head_hash: str | None = None |
| 183 | |
| 184 | @classmethod |
| 185 | def initialize(cls, *, mission_id: str, agent_id: str) -> AuditChain: |
| 186 | """ |
| 187 | Create a new AuditChain and write the genesis IDENTITY entry. |
| 188 | Call at Mission CREATED state. |
| 189 | """ |
| 190 | chain = cls(mission_id=mission_id, agent_id=agent_id) |
| 191 | genesis = AuditEntry.create( |
| 192 | chain_id=chain.chain_id, |
| 193 | sequence=0, |
| 194 | layer=AuditLayer.IDENTITY, |
| 195 | event="IDENTITY_chain_initialized", |
| 196 | agent_id=agent_id, |
| 197 | mission_id=mission_id, |
| 198 | prev_hash=_GENESIS_HASH, |
| 199 | data={"initialized_at": chain.initialized_at.isoformat()}, |
| 200 | ) |
| 201 | return chain.model_copy(update={ |
| 202 | "entries": [genesis], |
| 203 | "head_hash": genesis.entry_hash, |
| 204 | }) |
| 205 | |
| 206 | def append( |
| 207 | self, |
| 208 | *, |
| 209 | layer: AuditLayer, |
| 210 | event: str, |
| 211 | data: dict | None = None, |
| 212 | token_id: str | None = None, |
no outgoing calls
no test coverage detected