Coverage for node / src / stigmem_node / routes / decay.py: 89%

1"""Decay sweeper HTTP route — Phase 6 (spec §15) + async job path (spec §15.4). 

2 

3POST /v1/decay/sweep 

4 → 200 sync result, or 202 { job_id, status, estimated_s } when scope > threshold. 

5 

6GET /v1/decay/jobs/:job_id 

7 → 200 job status/result, or 404 if not found. 

8""" 

9 

10from __future__ import annotations 

11 

12from typing import Annotated, Any 

13 

14from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query 

15from fastapi.responses import JSONResponse 

16 

17from ..auth import Identity, resolve_identity 

18from ..db import db 

19from ..jobs import create_job, get_job, mark_done, mark_failed, mark_running 

20from ..lifecycle.decay import run_decay_sweep 

21from ..models.constants import VALID_SCOPES 

22from ..settings import settings 

23 

24router = APIRouter(prefix="/v1/decay", tags=["decay"]) 

25 

26 

27def _decay_job_worker( 

28 job_id: str, 

29 ttl_seconds: int | None, 

30 min_confidence: float | None, 

31 scope: str | None, 

32 dry_run: bool, 

33 tenant_id: str, 

34) -> None: 

35 """Background task: run decay sweep and update job status.""" 

36 mark_running(job_id) 

37 try: 

38 result = run_decay_sweep( 

39 ttl_seconds=ttl_seconds, 

40 min_confidence=min_confidence, 

41 scope=scope, 

42 dry_run=dry_run, 

43 tenant_id=tenant_id, 

44 ) 

45 mark_done(job_id, result) 

46 except Exception as exc: 

47 mark_failed(job_id, str(exc)) 

48 

49 

50@router.post("/sweep") 

51def decay_sweep( 

52 background_tasks: BackgroundTasks, 

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

54 dry_run: bool = Query(False, description="Report what would be decayed without writing"), 

55 scope: str | None = Query(None, description="Restrict sweep to one scope"), 

56 ttl_seconds: int | None = Query( 

57 None, ge=0, description="Expire non-expiring facts older than N seconds (0 = all)" 

58 ), 

59 min_confidence: float | None = Query( 

60 None, ge=0.0, le=1.0, description="Expire active facts below this confidence" 

61 ), 

62) -> Any: 

63 """Mark stale facts as expired. Cron-friendly one-shot sweeper. 

64 

65 Returns 200 synchronously for scopes ≤ threshold facts (Spec-X9-Decay-Semantics). 

66 Returns 202 with job_id for larger scopes; poll GET /v1/decay/jobs/:job_id. 

67 Note: dry_run is always synchronous per Spec-X9-Decay-Semantics. 

68 """ 

69 if not identity.can_write(): 69 ↛ 70line 69 didn't jump to line 70 because the condition on line 69 was never true

70 raise HTTPException(status_code=403, detail="write permission required") 

71 if scope is not None and scope not in VALID_SCOPES: 

72 raise HTTPException(status_code=400, detail=f"scope must be one of {VALID_SCOPES}") 

73 

74 # Dry-run is always synchronous (spec §15.4). 

75 if not dry_run: 

76 # Tenant-scope the sizing counts so the sync-vs-async branch is not a 

77 # cross-tenant fact-count oracle (audit decay sibling of H4). 

78 with db() as conn: 

79 if scope is not None: 

80 scope_count: int = conn.execute( 

81 "SELECT COUNT(*) FROM facts WHERE scope = ? AND tenant_id = ?", 

82 [scope, identity.tenant_id], 

83 ).fetchone()[0] 

84 else: 

85 scope_count = conn.execute( 

86 "SELECT COUNT(*) FROM facts WHERE tenant_id = ?", 

87 [identity.tenant_id], 

88 ).fetchone()[0] 

89 

90 if scope_count > settings.async_job_threshold: 

91 estimated_s = max(60, scope_count // 1_000) 

92 job_id = create_job("decay", scope, estimated_s, identity.tenant_id) 

93 background_tasks.add_task( 

94 _decay_job_worker, 

95 job_id, 

96 ttl_seconds, 

97 min_confidence, 

98 scope, 

99 dry_run, 

100 identity.tenant_id, 

101 ) 

102 return JSONResponse( 

103 status_code=202, 

104 content={"job_id": job_id, "status": "pending", "estimated_s": estimated_s}, 

105 ) 

106 

107 return run_decay_sweep( 

108 ttl_seconds=ttl_seconds, 

109 min_confidence=min_confidence, 

110 scope=scope, 

111 dry_run=dry_run, 

112 tenant_id=identity.tenant_id, 

113 ) 

114 

115 

116@router.get("/jobs/{job_id}") 

117def get_decay_job( 

118 job_id: str, 

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

120) -> Any: 

121 """Poll the status of an async decay job (Spec-X9-Decay-Semantics).""" 

122 if not identity.can_read(): 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true

123 raise HTTPException(status_code=403, detail="read permission required") 

124 job = get_job(job_id, job_type="decay", tenant_id=identity.tenant_id) 

125 if job is None: 

126 raise HTTPException(status_code=404, detail="job not found") 

127 return job