Coverage for node / src / stigmem_node / routes / instruction.py: 82%

1"""Lazy instruction discovery — spec §21 (Phase 10). 

2 

3Routes: 

4 GET /v1/agents/{agent_id}/boot-stub §21.8.1 MUST 

5 GET /v1/agents/{agent_id}/instruction-manifest §21.8.2 MUST 

6 PUT /v1/agents/{agent_id}/instruction-manifest §21.8.3 MUST 

7 POST /v1/agents/{agent_id}/recall-instruction §21.8.4 MUST 

8 POST /v1/instruction/audit §21.8.5 SHOULD 

9 GET /v1/agents/{agent_id}/instruction-manifest/coverage §21.8.6 SHOULD 

10""" 

11 

12from __future__ import annotations 

13 

14import json 

15import logging 

16import re 

17import secrets 

18import time 

19import uuid 

20from datetime import UTC, datetime 

21from typing import Annotated, Any 

22 

23import yaml 

24from fastapi import APIRouter, Depends, HTTPException, Response, status 

25from fastapi.responses import PlainTextResponse 

26 

27from ..auth import Identity, resolve_identity 

28from ..cid import compute_cid 

29from ..db import db 

30from ..models.instruction import ( 

31 AuditSubmitRequest, 

32 ManifestEntry, 

33 PublishManifestRequest, 

34 RecallInstructionRequest, 

35) 

36 

37logger = logging.getLogger("stigmem.instruction") 

38 

39router = APIRouter(tags=["instruction"]) 

40 

41# --------------------------------------------------------------------------- 

42# Constants 

43# --------------------------------------------------------------------------- 

44 

45_MANIFEST_TOKEN_LIMIT = 1000 

46_BOOT_STUB_TOKEN_LIMIT = 500 

47_GUARANTEE_LOAD_CAP = 5 

48_AUDIT_TOKEN_PREFIX = "audi_" # nosec B105 — audit token prefix, not a password 

49_AUDEVENT_PREFIX = "audevent_" 

50_AUDIT_TOKEN_TTL_S = 86_400 # 24 hours 

51 

52# Registered wake-reason enum values used for task_type validation. 

53# Extend this set when new wake reasons are added to the platform. 

54_KNOWN_WAKE_REASONS: frozenset[str] = frozenset( 

55 { 

56 "issue_assigned", 

57 "issue_commented", 

58 "issue_blockers_resolved", 

59 "issue_children_completed", 

60 "issue_comment_mentioned", 

61 "routine_fired", 

62 "approval_resolved", 

63 "manual", 

64 } 

65) 

66 

67_ADAPTER_PROFILES = {"paperclip-claude-code", "openai-assistants", "generic"} 

68 

69 

70# --------------------------------------------------------------------------- 

71# Helpers 

72# --------------------------------------------------------------------------- 

73 

74 

75def _approx_tokens(text: str) -> int: 

76 """Approximate cl100k token count (4 chars ≈ 1 token).""" 

77 try: 

78 import tiktoken 

79 

80 enc = tiktoken.get_encoding("cl100k_base") 

81 return len(enc.encode(text)) 

82 except Exception: 

83 return max(1, len(text) // 4) 

84 

85 

86def _now_ms() -> int: 

87 return int(time.time() * 1000) 

88 

89 

90def _is_admin(identity: Identity) -> bool: 

91 return identity.is_admin() 

92 

93 

94def _agent_uri_segments(entity_uri: str) -> set[str]: 

95 """Split an entity_uri into whole path/scheme segments. 

96 

97 Used so an agent_id must match a *complete* segment of the caller's 

98 entity_uri, never a substring (audit H3 / F-4): "cto" must not match 

99 "cto-shadow". 

100 """ 

101 return {seg for seg in re.split(r"[/:]+", entity_uri) if seg} 

102 

103 

104def _check_agent_access(identity: Identity, agent_id: str) -> None: 

105 """Raise 403 unless caller is the named agent or an admin.""" 

106 if _is_admin(identity): 

107 return 

108 # Agent key entity_uri must contain the agent_id as a WHOLE segment 

109 # (UUID or role slug), not merely a substring (audit H3 / F-4). 

110 if agent_id and agent_id in _agent_uri_segments(identity.entity_uri): 

111 return 

112 raise HTTPException( 

113 status_code=status.HTTP_403_FORBIDDEN, 

114 detail="instruction_scope_denied", 

115 ) 

116 

117 

118def _get_current_manifest(conn: Any, agent_id: str, tenant_id: str) -> dict[str, Any] | None: 

119 row = conn.execute( 

120 "SELECT * FROM instruction_manifests" 

121 " WHERE agent_id = ? AND tenant_id = ? AND superseded_at IS NULL" 

122 " ORDER BY created_at DESC LIMIT 1", 

123 (agent_id, tenant_id), 

124 ).fetchone() 

125 if row is None: 

126 return None 

127 return dict(row) 

128 

129 

130def _build_boot_stub( 

131 agent_id: str, 

132 agent_role: str, 

133 manifest_uri: str, 

134 manifest_version: str, 

135 adapter_profile: str, 

136 deployment: str = "default", 

137) -> str: 

138 frontmatter = { 

139 "agent_id": agent_id, 

140 "agent_role": agent_role, 

141 "heartbeat_contract": f"instruction:{deployment}/shared/heartbeat-contract/v1", 

142 "manifest_uri": manifest_uri, 

143 "stub_version": 1, 

144 "generated_at": datetime.now(UTC).isoformat(), 

145 "adapter_profile": adapter_profile, 

146 "migration_mode": "stigmem", 

147 } 

148 recall_schema = { 

149 "name": "recall_instruction", 

150 "description": "Retrieve relevant instruction units from the agent manifest.", 

151 "parameters": { 

152 "type": "object", 

153 "properties": { 

154 "intent": {"type": "string", "description": "What you are about to do"}, 

155 "max_chunks": {"type": "integer", "default": 3}, 

156 "token_budget": {"type": "integer", "default": _BOOT_STUB_TOKEN_LIMIT}, 

157 "manifest_hint": { 

158 "type": "array", 

159 "items": {"type": "string"}, 

160 "description": "Explicit unit names to prioritize", 

161 }, 

162 }, 

163 "required": ["intent"], 

164 }, 

165 } 

166 yaml_str = yaml.dump( 

167 {**frontmatter, "recall_tool_schema": recall_schema}, 

168 default_flow_style=False, 

169 allow_unicode=True, 

170 sort_keys=False, 

171 ) 

172 body = ( 

173 f"# Agent Boot Stub\n\n" 

174 f"You are **{agent_role}** (id: `{agent_id}`).\n\n" 

175 f"Your heartbeat procedure is at `{frontmatter['heartbeat_contract']}`.\n" 

176 f"Your instruction manifest is at `{manifest_uri}`.\n\n" 

177 f"Call `recall_instruction(intent)` to load relevant instruction sections before\n" 

178 f"performing any non-trivial task. The manifest lists available sections and their\n" 

179 f"triggers to help you decide when to load.\n" 

180 ) 

181 return f"---\n{yaml_str}---\n\n{body}" 

182 

183 

184def _validate_manifest_entries(entries: list[ManifestEntry]) -> None: 

185 seen_names: set[str] = set() 

186 guarantee_count = 0 

187 

188 for entry in entries: 

189 # Exactly one of fact_uri / path must be present 

190 has_fact = bool(entry.fact_uri) 

191 has_path = bool(entry.path) 

192 if not has_fact and not has_path: 

193 raise HTTPException( 

194 400, 

195 detail=(f"manifest_entry_invalid: '{entry.name}' has neither fact_uri nor path"), 

196 ) 

197 if has_fact and has_path: 

198 raise HTTPException( 

199 400, 

200 detail=(f"manifest_entry_invalid: '{entry.name}' has both fact_uri and path"), 

201 ) 

202 

203 # Unique names 

204 if entry.name in seen_names: 204 ↛ 205line 204 didn't jump to line 205 because the condition on line 204 was never true

205 raise HTTPException( 

206 400, 

207 detail=f"manifest_entry_invalid: duplicate name '{entry.name}'", 

208 ) 

209 seen_names.add(entry.name) 

210 

211 # Validate required_by_task_types 

212 for tt in entry.required_by_task_types: 

213 if tt not in _KNOWN_WAKE_REASONS: 

214 raise HTTPException( 

215 400, 

216 detail=f"task_type_unknown: '{tt}' is not a registered wake-reason", 

217 ) 

218 if len(entry.required_by_task_types) > 2: 

219 raise HTTPException( 

220 400, 

221 detail=( 

222 "task_types_approval_required: entry declares > 2 " 

223 "required_by_task_types; admin approval required" 

224 ), 

225 ) 

226 

227 if entry.guarantee_load: 

228 guarantee_count += 1 

229 

230 if guarantee_count > _GUARANTEE_LOAD_CAP: 

231 raise HTTPException( 

232 400, 

233 detail=( 

234 f"guarantee_cap_exceeded: at most {_GUARANTEE_LOAD_CAP} entries " 

235 "may have guarantee_load=true per agent" 

236 ), 

237 ) 

238 

239 

240def _score_intent_against_entry(intent: str, entry: ManifestEntry) -> float: 

241 """Simple BM25-style keyword overlap score for ranking manifest entries.""" 

242 intent_words = set(re.findall(r"\w+", intent.lower())) 

243 if not intent_words: 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true

244 return 0.0 

245 score = 0.0 

246 # Check description overlap 

247 desc_words = set(re.findall(r"\w+", entry.description.lower())) 

248 score += len(intent_words & desc_words) / max(len(intent_words), 1) * 0.4 

249 # Check trigger intents 

250 for trigger_intent in entry.load_triggers.intents: 

251 trigger_words = set(re.findall(r"\w+", trigger_intent.lower())) 

252 score += len(intent_words & trigger_words) / max(len(intent_words), 1) * 0.3 

253 # Check keywords 

254 for kw in entry.load_triggers.keywords: 

255 if kw.lower() in intent.lower(): 

256 score += 0.2 

257 return min(score, 1.0) 

258 

259 

260def _fetch_instruction_content(entry: ManifestEntry, tenant_id: str) -> tuple[str, str]: 

261 """Return (content, source) for a manifest entry. Raises on failure. 

262 

263 Scoped by tenant (audit H2): an instruction fact must resolve only within 

264 the caller's tenant, else another tenant's fact at the same entity URI 

265 leaks into — and steers — this agent's instruction channel. 

266 """ 

267 if entry.fact_uri: 267 ↛ 276line 267 didn't jump to line 276 because the condition on line 267 was always true

268 with db() as conn: 

269 row = conn.execute( 

270 "SELECT value_v, valid_until FROM facts" 

271 " WHERE entity = ? AND tenant_id = ? ORDER BY timestamp DESC LIMIT 1", 

272 (entry.fact_uri, tenant_id), 

273 ).fetchone() 

274 if row: 274 ↛ 276line 274 didn't jump to line 276

275 return str(row["value_v"]), "stigmem" 

276 if entry.path: 

277 try: 

278 with open(entry.path) as f: 

279 return f.read(), "fallback_path" 

280 except OSError as exc: 

281 raise LookupError( 

282 f"instruction fallback path '{entry.path}' could not be read" 

283 ) from exc 

284 raise LookupError(f"instruction content not found for entry '{entry.name}'") 

285 

286 

287def _get_fact_valid_until(fact_uri: str, tenant_id: str) -> str | None: 

288 with db() as conn: 

289 row = conn.execute( 

290 "SELECT valid_until FROM facts WHERE entity = ? AND tenant_id = ?" 

291 " ORDER BY timestamp DESC LIMIT 1", 

292 (fact_uri, tenant_id), 

293 ).fetchone() 

294 if row: 294 ↛ 297line 294 didn't jump to line 297 because the condition on line 294 was always true

295 valid_until: str | None = row["valid_until"] 

296 return valid_until 

297 return None 

298 

299 

300# --------------------------------------------------------------------------- 

301# 21.8.1 Get Boot Stub 

302# --------------------------------------------------------------------------- 

303 

304 

305@router.get("/v1/agents/{agent_id}/boot-stub", response_class=PlainTextResponse) 

306def get_boot_stub( 

307 agent_id: str, 

308 identity: Annotated[Identity, Depends(resolve_identity)], 

309 profile: str = "generic", 

310) -> PlainTextResponse: 

311 _check_agent_access(identity, agent_id) 

312 

313 if profile not in _ADAPTER_PROFILES: 

314 profile = "generic" 

315 

316 with db() as conn: 

317 stub_row = conn.execute( 

318 "SELECT body, token_count, manifest_version FROM boot_stubs" 

319 " WHERE agent_id = ? AND adapter_profile = ? AND tenant_id = ?", 

320 (agent_id, profile, identity.tenant_id), 

321 ).fetchone() 

322 

323 if stub_row is None: 323 ↛ 370line 323 didn't jump to line 370 because the condition on line 323 was always true

324 manifest_row = _get_current_manifest(conn, agent_id, identity.tenant_id) 

325 if manifest_row is None: 325 ↛ 326line 325 didn't jump to line 326 because the condition on line 325 was never true

326 raise HTTPException( 

327 status_code=status.HTTP_404_NOT_FOUND, 

328 detail="boot_stub_not_found", 

329 ) 

330 # Generate on the fly from manifest 

331 agent_role = _derive_agent_role(agent_id, conn) 

332 manifest_uri = manifest_row["fact_uri"] 

333 stub_body = _build_boot_stub( 

334 agent_id=agent_id, 

335 agent_role=agent_role, 

336 manifest_uri=manifest_uri, 

337 manifest_version=manifest_row["version"], 

338 adapter_profile=profile, 

339 ) 

340 token_count = _approx_tokens(stub_body) 

341 if token_count > _BOOT_STUB_TOKEN_LIMIT: 341 ↛ 342line 341 didn't jump to line 342 because the condition on line 341 was never true

342 raise HTTPException( 

343 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 

344 detail=( 

345 f"boot_stub_too_large: {token_count} tokens exceeds " 

346 f"{_BOOT_STUB_TOKEN_LIMIT}" 

347 ), 

348 ) 

349 now_ms = _now_ms() 

350 conn.execute( 

351 """INSERT OR REPLACE INTO boot_stubs 

352 (agent_id, adapter_profile, stub_version, body, token_count, 

353 generated_at, manifest_version, tenant_id) 

354 VALUES (?,?,?,?,?,?,?,?)""", 

355 ( 

356 agent_id, 

357 profile, 

358 1, 

359 stub_body, 

360 token_count, 

361 now_ms, 

362 manifest_row["version"], 

363 identity.tenant_id, 

364 ), 

365 ) 

366 stub_body_out = stub_body 

367 token_count_out = token_count 

368 manifest_version_out = manifest_row["version"] 

369 else: 

370 stub_body_out = stub_row["body"] 

371 token_count_out = stub_row["token_count"] 

372 manifest_version_out = stub_row["manifest_version"] 

373 

374 return PlainTextResponse( 

375 content=stub_body_out, 

376 media_type="text/markdown", 

377 headers={ 

378 "X-Stub-Version": "1", 

379 "X-Manifest-Version": manifest_version_out, 

380 "X-Token-Count": str(token_count_out), 

381 }, 

382 ) 

383 

384 

385def _derive_agent_role(agent_id: str, conn: Any) -> str: 

386 """Best-effort: look up a human-readable role for agent_id.""" 

387 # Escape LIKE metacharacters so a caller-supplied agent_id cannot inject 

388 # wildcards (audit F-10): "%"/"_" must match literally, not every row. 

389 escaped = agent_id.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") 

390 row = conn.execute( 

391 "SELECT entity_uri FROM api_keys WHERE entity_uri LIKE ? ESCAPE '\\' LIMIT 1", 

392 (f"%{escaped}%",), 

393 ).fetchone() 

394 if row: 

395 uri: str = row["entity_uri"] 

396 # e.g. "agent:cto" or "stigmem://org/agent/cto" 

397 parts = uri.replace("//", "/").rstrip("/").split("/") 

398 if parts: 398 ↛ 400line 398 didn't jump to line 400 because the condition on line 398 was always true

399 return parts[-1].upper() 

400 return "Agent" 

401 

402 

403# --------------------------------------------------------------------------- 

404# 21.8.2 Get Instruction Manifest 

405# --------------------------------------------------------------------------- 

406 

407 

408@router.get("/v1/agents/{agent_id}/instruction-manifest") 

409def get_instruction_manifest( 

410 agent_id: str, 

411 identity: Annotated[Identity, Depends(resolve_identity)], 

412) -> dict[str, Any]: 

413 _check_agent_access(identity, agent_id) 

414 

415 with db() as conn: 

416 row = _get_current_manifest(conn, agent_id, identity.tenant_id) 

417 

418 if row is None: 418 ↛ 419line 418 didn't jump to line 419 because the condition on line 418 was never true

419 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="manifest_not_found") 

420 

421 entries = json.loads(row["body"]) 

422 created_ms: int = row["created_at"] 

423 last_updated = datetime.fromtimestamp(created_ms / 1000, tz=UTC).isoformat() 

424 

425 return { 

426 "manifest_version": row["version"], 

427 "fact_uri": row["fact_uri"], 

428 "token_count": row["token_count"], 

429 "entries": entries, 

430 "last_updated_at": last_updated, 

431 } 

432 

433 

434# --------------------------------------------------------------------------- 

435# 21.8.3 Publish / Replace Instruction Manifest 

436# --------------------------------------------------------------------------- 

437 

438 

439@router.put("/v1/agents/{agent_id}/instruction-manifest", status_code=200) 

440def publish_instruction_manifest( 

441 agent_id: str, 

442 req: PublishManifestRequest, 

443 identity: Annotated[Identity, Depends(resolve_identity)], 

444) -> dict[str, Any]: 

445 if not _is_admin(identity): 

446 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="admin key required") 

447 

448 _validate_manifest_entries(req.entries) 

449 

450 # Serialize entries for storage 

451 entries_json = json.dumps([e.model_dump() for e in req.entries]) 

452 token_count = _approx_tokens(entries_json) 

453 if token_count > _MANIFEST_TOKEN_LIMIT: 453 ↛ 454line 453 didn't jump to line 454 because the condition on line 453 was never true

454 raise HTTPException( 

455 400, 

456 detail=(f"manifest_too_large: {token_count} tokens exceeds {_MANIFEST_TOKEN_LIMIT}"), 

457 ) 

458 

459 # Build the instruction: URI for this manifest 

460 fact_uri = f"instruction:default/agent/{agent_id}/manifest/{req.version}" 

461 

462 with db() as conn: 

463 # Check version uniqueness within the caller's tenant 

464 existing = conn.execute( 

465 "SELECT id FROM instruction_manifests" 

466 " WHERE agent_id = ? AND version = ? AND tenant_id = ?", 

467 (agent_id, req.version, identity.tenant_id), 

468 ).fetchone() 

469 if existing: 

470 raise HTTPException(status_code=409, detail="manifest_version_conflict") 

471 

472 # Run coverage gate (simplified: check entry validity; full paraphrase eval is Phase 11) 

473 coverage_report = [] 

474 for entry in req.entries: 

475 coverage_pct: float 

476 if req.skip_coverage_gate: 

477 coverage_pct = 1.0 

478 passed = True 

479 else: 

480 # Lightweight check: verify fact_uri exists if specified 

481 # (full N=5 paraphrase eval is Phase 11) 

482 if entry.fact_uri: 482 ↛ 492line 482 didn't jump to line 492 because the condition on line 482 was always true

483 row = conn.execute( 

484 "SELECT id FROM facts WHERE entity = ? AND tenant_id = ? LIMIT 1", 

485 (entry.fact_uri, identity.tenant_id), 

486 ).fetchone() 

487 # If fact doesn't exist yet (pre-seeding), warn but don't block 

488 coverage_pct = 1.0 if row else 0.5 

489 passed = coverage_pct >= 0.80 

490 else: 

491 # path-only entries pass coverage (read from filesystem, not stigmem) 

492 coverage_pct = 1.0 

493 passed = True 

494 coverage_report.append( 

495 { 

496 "unit": entry.name, 

497 "coverage_pct": coverage_pct, 

498 "passed": passed, 

499 } 

500 ) 

501 

502 failing = [r["unit"] for r in coverage_report if not r["passed"]] 

503 if failing and not req.skip_coverage_gate: 503 ↛ 504line 503 didn't jump to line 504 because the condition on line 503 was never true

504 raise HTTPException( 

505 status_code=400, 

506 detail=f"manifest_coverage_failure: units failed coverage gate: {failing}", 

507 ) 

508 

509 now_ms = _now_ms() 

510 manifest_id = str(uuid.uuid4()) 

511 

512 # Supersede previous current version (within this tenant) 

513 conn.execute( 

514 "UPDATE instruction_manifests SET superseded_at = ?" 

515 " WHERE agent_id = ? AND tenant_id = ? AND superseded_at IS NULL", 

516 (now_ms, agent_id, identity.tenant_id), 

517 ) 

518 

519 conn.execute( 

520 """INSERT INTO instruction_manifests 

521 (id, agent_id, version, fact_uri, token_count, body, created_at, tenant_id) 

522 VALUES (?,?,?,?,?,?,?,?)""", 

523 ( 

524 manifest_id, 

525 agent_id, 

526 req.version, 

527 fact_uri, 

528 token_count, 

529 entries_json, 

530 now_ms, 

531 identity.tenant_id, 

532 ), 

533 ) 

534 

535 # Invalidate boot stub cache for all profiles (within this tenant) 

536 conn.execute( 

537 "DELETE FROM boot_stubs WHERE agent_id = ? AND tenant_id = ?", 

538 (agent_id, identity.tenant_id), 

539 ) 

540 

541 # Store the manifest itself as a fact in the instruction: scope. 

542 # Stamp the caller's tenant, bind interpret_as, and persist a CID so the 

543 # manifest fact is tenant-isolated and read-path integrity-verifiable — 

544 # the raw insert previously omitted all three (audit M4). 

545 fact_id = str(uuid.uuid4()) 

546 ts = datetime.now(UTC).isoformat() 

547 manifest_cid = compute_cid( 

548 entity=fact_uri, 

549 relation="instruction:manifest", 

550 value_type="text", 

551 value_v=entries_json, 

552 source=identity.entity_uri, 

553 scope="local", 

554 confidence=1.0, 

555 interpret_as="instruction", 

556 ) 

557 conn.execute( 

558 """INSERT INTO facts 

559 (id, entity, relation, value_type, value_v, source, confidence, scope, 

560 timestamp, valid_until, garden_id, tenant_id, interpret_as, cid) 

561 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", 

562 ( 

563 fact_id, 

564 fact_uri, 

565 "instruction:manifest", 

566 "text", 

567 entries_json, 

568 identity.entity_uri, 

569 1.0, 

570 "local", 

571 ts, 

572 None, 

573 None, 

574 identity.tenant_id, 

575 "instruction", 

576 manifest_cid, 

577 ), 

578 ) 

579 

580 logger.info( 

581 "Instruction manifest published: agent=%s version=%s units=%d", 

582 agent_id, 

583 req.version, 

584 len(req.entries), 

585 ) 

586 

587 return { 

588 "fact_uri": fact_uri, 

589 "token_count": token_count, 

590 "coverage_report": coverage_report, 

591 } 

592 

593 

594# --------------------------------------------------------------------------- 

595# 21.8.4 Recall Instructions 

596# --------------------------------------------------------------------------- 

597 

598 

599def _resolve_hint_chunks( 

600 entries: list[ManifestEntry], 

601 hints: list[str], 

602 token_budget: int, 

603 tenant_id: str, 

604) -> tuple[list[dict[str, Any]], set[str], list[str], int]: 

605 """Step 1: resolve manifest_hint entries. 

606 

607 Returns (chunks, used_names, missed_hints, tokens_used). 

608 """ 

609 chunks: list[dict[str, Any]] = [] 

610 used_names: set[str] = set() 

611 missed_hints: list[str] = [] 

612 tokens_used = 0 

613 for hint_name in hints: 

614 entry = next((e for e in entries if e.name == hint_name), None) 

615 if entry is None: 

616 missed_hints.append(hint_name) 

617 continue 

618 try: 

619 content, source = _fetch_instruction_content(entry, tenant_id) 

620 except LookupError: 

621 missed_hints.append(hint_name) 

622 continue 

623 tokens = _approx_tokens(content) 

624 if tokens_used + tokens <= token_budget: 624 ↛ 613line 624 didn't jump to line 613 because the condition on line 624 was always true

625 chunks.append( 

626 _make_chunk(entry, content, tokens, source, score=1.0, tenant_id=tenant_id) 

627 ) 

628 used_names.add(entry.name) 

629 tokens_used += tokens 

630 return chunks, used_names, missed_hints, tokens_used 

631 

632 

633def _resolve_ranked_chunks( 

634 entries: list[ManifestEntry], 

635 intent: str, 

636 used_names: set[str], 

637 remaining_slots: int, 

638 token_budget: int, 

639 tokens_used: int, 

640 tenant_id: str, 

641) -> tuple[list[dict[str, Any]], int]: 

642 """Step 2: ranked retrieval for remaining slots. Returns (new_chunks, updated_tokens_used).""" 

643 if remaining_slots <= 0: 643 ↛ 644line 643 didn't jump to line 644 because the condition on line 643 was never true

644 return [], tokens_used 

645 scored: list[tuple[float, ManifestEntry]] = [] 

646 for entry in entries: 

647 if entry.name in used_names or entry.guarantee_load: 

648 continue # guaranteed entries handled in step 3 

649 scored.append((_score_intent_against_entry(intent, entry), entry)) 

650 scored.sort(key=lambda x: -x[0]) 

651 

652 new_chunks: list[dict[str, Any]] = [] 

653 for score, entry in scored[:remaining_slots]: 

654 try: 

655 content, source = _fetch_instruction_content(entry, tenant_id) 

656 except LookupError as exc: 

657 logger.debug("skipping recall candidate %s: %s", entry.name, exc) 

658 continue 

659 tokens = _approx_tokens(content) 

660 if tokens_used + tokens <= token_budget: 660 ↛ 653line 660 didn't jump to line 653 because the condition on line 660 was always true

661 new_chunks.append( 

662 _make_chunk(entry, content, tokens, source, score=score, tenant_id=tenant_id) 

663 ) 

664 used_names.add(entry.name) 

665 tokens_used += tokens 

666 return new_chunks, tokens_used 

667 

668 

669def _append_guaranteed_chunks( 

670 entries: list[ManifestEntry], 

671 used_names: set[str], 

672 chunks: list[dict[str, Any]], 

673 tokens_used: int, 

674 token_budget: int, 

675 tenant_id: str, 

676) -> tuple[int, bool]: 

677 """Step 3: insert/append guaranteed entries. Returns (tokens_used, truncated_flag).""" 

678 truncated = False 

679 guaranteed = [e for e in entries if e.guarantee_load and e.name not in used_names] 

680 

681 for entry in [e for e in guaranteed if e.force_position == "prepend"]: 681 ↛ 682line 681 didn't jump to line 682 because the loop on line 681 never started

682 try: 

683 content, source = _fetch_instruction_content(entry, tenant_id) 

684 except LookupError as exc: 

685 logger.warning("guaranteed prepend instruction %s unavailable: %s", entry.name, exc) 

686 continue 

687 tokens = _approx_tokens(content) 

688 chunks.insert( 

689 0, _make_chunk(entry, content, tokens, source, score=1.0, tenant_id=tenant_id) 

690 ) 

691 used_names.add(entry.name) 

692 tokens_used += tokens 

693 if tokens_used > token_budget: 

694 truncated = True 

695 

696 for entry in [e for e in guaranteed if e.force_position != "prepend"]: 696 ↛ 697line 696 didn't jump to line 697 because the loop on line 696 never started

697 try: 

698 content, source = _fetch_instruction_content(entry, tenant_id) 

699 except LookupError as exc: 

700 logger.warning("guaranteed append instruction %s unavailable: %s", entry.name, exc) 

701 continue 

702 tokens = _approx_tokens(content) 

703 chunks.append(_make_chunk(entry, content, tokens, source, score=1.0, tenant_id=tenant_id)) 

704 used_names.add(entry.name) 

705 tokens_used += tokens 

706 if tokens_used > token_budget: 

707 truncated = True 

708 

709 return tokens_used, truncated 

710 

711 

712def _write_recall_audit( 

713 audit_id: str, 

714 agent_id: str, 

715 identity: Identity, 

716 intent: str, 

717 loaded_chunk_names: list[str], 

718 audit_token: str, 

719 now_ms: int, 

720) -> None: 

721 """Best-effort INSERT into instruction_audit; failures are logged not raised.""" 

722 try: 

723 with db() as conn: 

724 conn.execute( 

725 """INSERT INTO instruction_audit 

726 (id, agent_id, heartbeat_id, session_start, intent, loaded_chunks, 

727 used_chunks, missed_chunks, audit_token, audit_closed, created_at, tenant_id) 

728 VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""", 

729 ( 

730 audit_id, 

731 agent_id, 

732 identity.entity_uri, 

733 now_ms, 

734 intent, 

735 json.dumps(loaded_chunk_names), 

736 "[]", 

737 "[]", 

738 audit_token, 

739 None, 

740 now_ms, 

741 identity.tenant_id, 

742 ), 

743 ) 

744 except Exception as exc: 

745 logger.warning("audit_write_failed: %s", exc) 

746 

747 

748@router.post("/v1/agents/{agent_id}/recall-instruction") 

749def recall_instruction( 

750 agent_id: str, 

751 req: RecallInstructionRequest, 

752 identity: Annotated[Identity, Depends(resolve_identity)], 

753) -> dict[str, Any]: 

754 _check_agent_access(identity, agent_id) 

755 

756 with db() as conn: 

757 manifest_row = _get_current_manifest(conn, agent_id, identity.tenant_id) 

758 if manifest_row is None: 

759 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="manifest_not_found") 

760 

761 entries: list[ManifestEntry] = [ManifestEntry(**e) for e in json.loads(manifest_row["body"])] 

762 

763 chunks, used_names, missed_hints, tokens_used = _resolve_hint_chunks( 

764 entries, 

765 req.manifest_hint, 

766 req.token_budget, 

767 identity.tenant_id, 

768 ) 

769 

770 ranked_chunks, tokens_used = _resolve_ranked_chunks( 

771 entries, 

772 req.intent, 

773 used_names, 

774 req.max_chunks - len(chunks), 

775 req.token_budget, 

776 tokens_used, 

777 identity.tenant_id, 

778 ) 

779 chunks.extend(ranked_chunks) 

780 

781 tokens_used, truncated = _append_guaranteed_chunks( 

782 entries, 

783 used_names, 

784 chunks, 

785 tokens_used, 

786 req.token_budget, 

787 identity.tenant_id, 

788 ) 

789 

790 audit_token = _AUDIT_TOKEN_PREFIX + secrets.token_urlsafe(16) 

791 now_ms = _now_ms() 

792 _write_recall_audit( 

793 _AUDEVENT_PREFIX + str(uuid.uuid4()), 

794 agent_id, 

795 identity, 

796 req.intent, 

797 [c["name"] for c in chunks], 

798 audit_token, 

799 now_ms, 

800 ) 

801 

802 return { 

803 "chunks": chunks, 

804 "total_tokens": tokens_used, 

805 "truncated": truncated, 

806 "missed_hints": missed_hints, 

807 "audit_token": audit_token, 

808 } 

809 

810 

811def _make_chunk( 

812 entry: ManifestEntry, content: str, tokens: int, source: str, score: float, tenant_id: str 

813) -> dict[str, Any]: 

814 valid_until = _get_fact_valid_until(entry.fact_uri, tenant_id) if entry.fact_uri else None 

815 # Extract version from fact_uri, e.g. "instruction:.../v2" → "v2" 

816 version = "v1" 

817 if entry.fact_uri: 817 ↛ 821line 817 didn't jump to line 821 because the condition on line 817 was always true

818 parts = entry.fact_uri.rstrip("/").split("/") 

819 if parts: 819 ↛ 821line 819 didn't jump to line 821 because the condition on line 819 was always true

820 version = parts[-1] 

821 return { 

822 "name": entry.name, 

823 "fact_uri": entry.fact_uri, 

824 "content": content, 

825 "tokens": tokens, 

826 "valid_until": valid_until, 

827 "version": version, 

828 "score": round(score, 4), 

829 "source": source, 

830 } 

831 

832 

833# --------------------------------------------------------------------------- 

834# 21.8.5 Submit Discovery Audit 

835# --------------------------------------------------------------------------- 

836 

837 

838@router.post("/v1/instruction/audit", status_code=status.HTTP_204_NO_CONTENT) 

839def submit_discovery_audit( 

840 req: AuditSubmitRequest, 

841 identity: Annotated[Identity, Depends(resolve_identity)], 

842) -> Response: 

843 now_ms = _now_ms() 

844 

845 with db() as conn: 

846 row = conn.execute( 

847 "SELECT id, agent_id, created_at, audit_closed FROM instruction_audit" 

848 " WHERE audit_token = ? AND tenant_id = ?", 

849 (req.audit_token, identity.tenant_id), 

850 ).fetchone() 

851 

852 if row is None: 

853 raise HTTPException(400, detail="audit_token_invalid") 

854 

855 # Bind the submitter to the audit's agent (audit F-8): possession of 

856 # the token is necessary but not sufficient — the caller must be that 

857 # agent (or an admin), else a leaked token closes another agent's audit. 

858 _check_agent_access(identity, row["agent_id"]) 

859 

860 # Idempotent: already closed 

861 if row["audit_closed"] is not None: 

862 return Response(status_code=status.HTTP_204_NO_CONTENT) 

863 

864 # TTL check 

865 age_s = (now_ms - row["created_at"]) / 1000 

866 if age_s > _AUDIT_TOKEN_TTL_S: 866 ↛ 867line 866 didn't jump to line 867 because the condition on line 866 was never true

867 raise HTTPException(400, detail="audit_token_expired") 

868 

869 conn.execute( 

870 "UPDATE instruction_audit" 

871 " SET used_chunks = ?, missed_chunks = ?, audit_closed = ?" 

872 " WHERE audit_token = ? AND tenant_id = ?", 

873 ( 

874 json.dumps(req.used_chunks),