personal memory agent
think/surfaces/health: add V1 consumer-surface Health with structural-note contract
Adds read-only meta-surface reporting journal trust state: capture
coverage, synthesis completeness, consumer-signal health. Emits explicit
HealthNotes for missing signals and stale state rather than silently
zeroing.
- think/surfaces/types.py: append 5 frozen dataclasses per LOCKED shape
(CaptureHealth, SynthesisHealth, ConsumerSignalHealth, HealthNote,
HealthReport). Field names/types/order fixed by scope contract.
- think/surfaces/health.py: summary/full/for_range returning HealthReport.
One shared range scan across get_facets() × days feeds capture and
synthesis builders; consumer-signals composes ledger.list() twice.
Silent-facet ladder emits at most one note per facet (highest
severity); stale/missing indexer and missing talent day-index emit
dedicated notes. Two structural info notes (coverage_ratio and
corrections-ledger) always emitted per scope.
- think/tools/health.py: Typer sub-app mounting summary / full /
for-range / pipeline under sol call health. Pipeline subcommand
relocated verbatim from apps/health/call.py (local-time default
preserved). Top-level help disambiguates from sol health (service
liveness).
- think/call.py: re-alphabetize built-in import/mount block; register
health_app once.
- apps/health/call.py + apps/health/tests/test_call.py deleted to free
the sol call health namespace for the new built-in. Web surface
(routes.py, workspace.html, app.json, talent/health/) left intact.
- tests/test_surfaces_health.py: 23 tests covering every LOCKED metric,
note-emission rule, range validation, deterministic sort order, CLI
JSON round-trip, and pipeline subcommand behavior preservation.
Note sort order: (severity_rank, category, message) with
critical=0/warn=1/info=2, applied once in _build_report. Never silently
zero a missing signal. coverage_ratio always None in V1; hours_total
denominator shipped, but the ratio is withheld until Sprint 5+ per
cpo/specs/in-flight/consumer-surface-health.md.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
+1684
-124
-41
apps/health/call.py
-41
apps/health/call.py
···
1
-
# SPDX-License-Identifier: AGPL-3.0-only
2
-
# Copyright (c) 2026 sol pbc
3
-
4
-
"""Health diagnostics — exposes `sol call health <command>` entry points."""
5
-
6
-
from __future__ import annotations
7
-
8
-
import json
9
-
from datetime import datetime, timedelta
10
-
from typing import Optional
11
-
12
-
import typer
13
-
14
-
from think.pipeline_health import summarize_pipeline_day
15
-
16
-
app = typer.Typer(help="Health diagnostics — sol call health ...")
17
-
18
-
19
-
@app.command("pipeline")
20
-
def pipeline(
21
-
day: Optional[str] = typer.Option(
22
-
None, "--day", help="Day to summarize (YYYYMMDD)."
23
-
),
24
-
yesterday: bool = typer.Option(
25
-
False, "--yesterday", help="Summarize yesterday's pipeline."
26
-
),
27
-
) -> None:
28
-
"""Summarize think pipeline health for one day."""
29
-
if day is not None and yesterday:
30
-
typer.echo("--day and --yesterday are mutually exclusive", err=True)
31
-
raise typer.Exit(1)
32
-
33
-
if day is not None:
34
-
target = day
35
-
elif yesterday:
36
-
target = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
37
-
else:
38
-
target = datetime.now().strftime("%Y%m%d")
39
-
40
-
summary = summarize_pipeline_day(target)
41
-
typer.echo(json.dumps(summary, indent=2, sort_keys=False))
-82
apps/health/tests/test_call.py
-82
apps/health/tests/test_call.py
···
1
-
# SPDX-License-Identifier: AGPL-3.0-only
2
-
# Copyright (c) 2026 sol pbc
3
-
4
-
"""Tests for health app call commands."""
5
-
6
-
from __future__ import annotations
7
-
8
-
import json
9
-
from datetime import datetime, timedelta
10
-
11
-
from typer.testing import CliRunner
12
-
13
-
from apps.health.call import app
14
-
15
-
runner = CliRunner()
16
-
17
-
18
-
def test_pipeline_default_is_today(health_env):
19
-
health_env()
20
-
21
-
result = runner.invoke(app, [])
22
-
23
-
assert result.exit_code == 0
24
-
payload = json.loads(result.stdout)
25
-
assert payload["day"] == datetime.now().strftime("%Y%m%d")
26
-
27
-
28
-
def test_pipeline_day_option(health_env):
29
-
health_env()
30
-
31
-
result = runner.invoke(app, ["--day", "20250101"])
32
-
33
-
assert result.exit_code == 0
34
-
payload = json.loads(result.stdout)
35
-
assert payload["day"] == "20250101"
36
-
37
-
38
-
def test_pipeline_yesterday_option(health_env):
39
-
health_env()
40
-
41
-
result = runner.invoke(app, ["--yesterday"])
42
-
43
-
assert result.exit_code == 0
44
-
payload = json.loads(result.stdout)
45
-
assert payload["day"] == (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
46
-
47
-
48
-
def test_mutual_exclusion_error(health_env):
49
-
health_env()
50
-
51
-
result = runner.invoke(app, ["--day", "20250101", "--yesterday"])
52
-
53
-
assert result.exit_code == 1
54
-
assert "mutually exclusive" in result.stderr
55
-
56
-
57
-
def test_pipeline_with_real_fixture(health_env):
58
-
env = health_env()
59
-
day = "20260101"
60
-
health_path = env.journal / day / "health" / "123_segment.jsonl"
61
-
health_path.parent.mkdir(parents=True, exist_ok=True)
62
-
health_path.write_text(
63
-
"\n".join(
64
-
[
65
-
json.dumps({"event": "run.start", "mode": "segment"}),
66
-
json.dumps({"event": "talent.dispatch", "mode": "segment"}),
67
-
json.dumps({"event": "talent.complete", "mode": "segment"}),
68
-
json.dumps(
69
-
{"event": "run.complete", "mode": "segment", "duration_ms": 42}
70
-
),
71
-
]
72
-
)
73
-
+ "\n",
74
-
encoding="utf-8",
75
-
)
76
-
77
-
result = runner.invoke(app, ["--day", day])
78
-
79
-
assert result.exit_code == 0
80
-
payload = json.loads(result.stdout)
81
-
assert payload["runs"]["segment"]["count"] == 1
82
-
assert payload["agents"]["dispatched"] >= 1
+933
tests/test_surfaces_health.py
+933
tests/test_surfaces_health.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
from __future__ import annotations
5
+
6
+
import json
7
+
import os
8
+
import re
9
+
from datetime import UTC, datetime, timedelta
10
+
from pathlib import Path
11
+
12
+
import pytest
13
+
from typer.testing import CliRunner
14
+
15
+
from think.surfaces import health as health_surface
16
+
17
+
_RUNNER = CliRunner()
18
+
_SPEC_POINTER = "cpo/specs/in-flight/consumer-surface-health.md"
19
+
20
+
21
+
def _configure_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
22
+
monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path))
23
+
monkeypatch.setenv("SOL_SKIP_SUPERVISOR_CHECK", "1")
24
+
25
+
from think.entities.journal import clear_journal_entity_cache
26
+
from think.entities.loading import clear_entity_loading_cache
27
+
from think.entities.relationships import clear_relationship_caches
28
+
29
+
clear_journal_entity_cache()
30
+
clear_entity_loading_cache()
31
+
clear_relationship_caches()
32
+
33
+
34
+
def _set_now(monkeypatch: pytest.MonkeyPatch, value: datetime) -> None:
35
+
assert value.tzinfo == UTC
36
+
monkeypatch.setattr(health_surface, "_resolve_now", lambda: value)
37
+
38
+
39
+
def _minimal_facet_tree(
40
+
tmp_path: Path,
41
+
*,
42
+
facets: tuple[str, ...] = ("work",),
43
+
muted_facets: tuple[str, ...] = (),
44
+
) -> None:
45
+
muted = set(muted_facets)
46
+
for facet in facets:
47
+
facet_dir = tmp_path / "facets" / facet
48
+
facet_dir.mkdir(parents=True, exist_ok=True)
49
+
(facet_dir / "activities").mkdir(exist_ok=True)
50
+
(facet_dir / "facet.json").write_text(
51
+
json.dumps(
52
+
{
53
+
"title": facet.title(),
54
+
"description": "",
55
+
"color": "",
56
+
"emoji": "",
57
+
"muted": facet in muted,
58
+
}
59
+
),
60
+
encoding="utf-8",
61
+
)
62
+
63
+
64
+
def _write_entity(
65
+
tmp_path: Path,
66
+
entity_id: str,
67
+
name: str,
68
+
*,
69
+
entity_type: str = "Person",
70
+
) -> None:
71
+
entity_dir = tmp_path / "entities" / entity_id
72
+
entity_dir.mkdir(parents=True, exist_ok=True)
73
+
(entity_dir / "entity.json").write_text(
74
+
json.dumps({"id": entity_id, "name": name, "type": entity_type}),
75
+
encoding="utf-8",
76
+
)
77
+
78
+
79
+
def _utc_dt(day: str, hour: int = 12, minute: int = 0) -> datetime:
80
+
return datetime.strptime(
81
+
f"{day} {hour:02d}:{minute:02d}:00",
82
+
"%Y%m%d %H:%M:%S",
83
+
).replace(tzinfo=UTC)
84
+
85
+
86
+
def _utc_ms(day: str, hour: int = 12, minute: int = 0) -> int:
87
+
return int(_utc_dt(day, hour, minute).timestamp() * 1000)
88
+
89
+
90
+
def _iso_utc(day: str, hour: int = 12, minute: int = 0) -> str:
91
+
return _utc_dt(day, hour, minute).isoformat().replace("+00:00", "Z")
92
+
93
+
94
+
def _activity_record(
95
+
day: str,
96
+
record_id: str,
97
+
*,
98
+
activity: str = "meeting",
99
+
segments: list[str] | None = None,
100
+
participation: object = None,
101
+
story: object = None,
102
+
edits: list[dict[str, object]] | None = None,
103
+
source: str = "user",
104
+
hidden: bool = False,
105
+
created_at: int | None = None,
106
+
start: str | None = None,
107
+
cancelled: bool = False,
108
+
commitments: list[dict[str, object]] | None = None,
109
+
closures: list[dict[str, object]] | None = None,
110
+
) -> dict[str, object]:
111
+
record: dict[str, object] = {
112
+
"id": record_id,
113
+
"activity": activity,
114
+
"title": record_id,
115
+
"description": record_id,
116
+
"segments": segments or [],
117
+
"created_at": created_at if created_at is not None else _utc_ms(day),
118
+
"source": source,
119
+
"hidden": hidden,
120
+
"edits": edits or [],
121
+
}
122
+
if participation is not None:
123
+
record["participation"] = participation
124
+
if story is not None:
125
+
record["story"] = story
126
+
if start is not None:
127
+
record["start"] = start
128
+
if cancelled:
129
+
record["cancelled"] = True
130
+
if commitments is not None:
131
+
record["commitments"] = commitments
132
+
if closures is not None:
133
+
record["closures"] = closures
134
+
return record
135
+
136
+
137
+
def _append_jsonl(path: Path, payload: dict[str, object]) -> None:
138
+
path.parent.mkdir(parents=True, exist_ok=True)
139
+
with path.open("a", encoding="utf-8") as handle:
140
+
handle.write(json.dumps(payload) + "\n")
141
+
142
+
143
+
def _write_activity(
144
+
tmp_path: Path,
145
+
facet: str,
146
+
day: str,
147
+
payload: dict[str, object],
148
+
) -> None:
149
+
_append_jsonl(tmp_path / "facets" / facet / "activities" / f"{day}.jsonl", payload)
150
+
151
+
152
+
def _write_talent_day(
153
+
tmp_path: Path,
154
+
day: str,
155
+
*rows: dict[str, object],
156
+
) -> None:
157
+
path = tmp_path / "talents" / f"{day}.jsonl"
158
+
path.parent.mkdir(parents=True, exist_ok=True)
159
+
path.write_text(
160
+
"".join(json.dumps(row) + "\n" for row in rows),
161
+
encoding="utf-8",
162
+
)
163
+
164
+
165
+
def _write_indexer_db(tmp_path: Path, dt: datetime) -> int:
166
+
path = tmp_path / "indexer" / "journal.sqlite"
167
+
path.parent.mkdir(parents=True, exist_ok=True)
168
+
path.write_text("", encoding="utf-8")
169
+
ts = dt.timestamp()
170
+
os.utime(path, (ts, ts))
171
+
return path.stat().st_mtime_ns // 1_000_000
172
+
173
+
174
+
def _write_pipeline_log(
175
+
tmp_path: Path,
176
+
day: str,
177
+
filename: str,
178
+
*rows: dict[str, object],
179
+
) -> None:
180
+
path = tmp_path / "chronicle" / day / "health" / filename
181
+
path.parent.mkdir(parents=True, exist_ok=True)
182
+
path.write_text(
183
+
"".join(json.dumps(row) + "\n" for row in rows),
184
+
encoding="utf-8",
185
+
)
186
+
187
+
188
+
def test_summary_metrics_from_controlled_range(tmp_path, monkeypatch):
189
+
_configure_env(tmp_path, monkeypatch)
190
+
_set_now(monkeypatch, _utc_dt("20260410"))
191
+
_minimal_facet_tree(tmp_path)
192
+
indexer_ms = _write_indexer_db(tmp_path, _utc_dt("20260410", 11, 0))
193
+
_write_talent_day(
194
+
tmp_path,
195
+
"20260410",
196
+
{
197
+
"use_id": "1",
198
+
"name": "flow",
199
+
"day": "20260410",
200
+
"facet": None,
201
+
"ts": _utc_ms("20260410", 9),
202
+
"status": "failed",
203
+
},
204
+
{
205
+
"use_id": "2",
206
+
"name": "flow",
207
+
"day": "20260410",
208
+
"facet": None,
209
+
"ts": _utc_ms("20260410", 8),
210
+
"status": "completed",
211
+
},
212
+
)
213
+
_write_talent_day(
214
+
tmp_path,
215
+
"20260409",
216
+
{
217
+
"use_id": "3",
218
+
"name": "flow",
219
+
"day": "20260409",
220
+
"facet": None,
221
+
"ts": _utc_ms("20260409", 15),
222
+
"status": "completed",
223
+
},
224
+
)
225
+
_write_activity(
226
+
tmp_path,
227
+
"work",
228
+
"20260409",
229
+
_activity_record(
230
+
"20260409",
231
+
"meeting_100000_3600",
232
+
segments=["100000_3600"],
233
+
participation=[{"entity_id": "alex"}],
234
+
story={"body": "Discussed launch."},
235
+
edits=[{"actor": "cli:update", "fields": ["details"]}],
236
+
created_at=_utc_ms("20260409", 10),
237
+
),
238
+
)
239
+
_write_activity(
240
+
tmp_path,
241
+
"work",
242
+
"20260410",
243
+
_activity_record(
244
+
"20260410",
245
+
"meeting_140000_1800",
246
+
segments=["140000_1800"],
247
+
edits=[{"actor": "system:story", "fields": ["story"]}],
248
+
created_at=_utc_ms("20260410", 14),
249
+
),
250
+
)
251
+
_write_activity(
252
+
tmp_path,
253
+
"work",
254
+
"20260410",
255
+
_activity_record(
256
+
"20260410",
257
+
"anticipated_090000_1800",
258
+
source="anticipated",
259
+
start=_iso_utc("20260410", 9),
260
+
created_at=_utc_ms("20260410", 8, 30),
261
+
),
262
+
)
263
+
_write_activity(
264
+
tmp_path,
265
+
"work",
266
+
"20260410",
267
+
_activity_record(
268
+
"20260410",
269
+
"hidden_080000_1800",
270
+
hidden=True,
271
+
participation=[{"entity_id": "ignored"}],
272
+
story={"body": "Ignored."},
273
+
edits=[{"actor": "cli:update", "fields": ["details"]}],
274
+
created_at=_utc_ms("20260410", 8),
275
+
),
276
+
)
277
+
278
+
report = health_surface.for_range("20260409", "20260410")
279
+
280
+
assert report.range == ("20260409", "20260410")
281
+
assert report.capture_health.hours_with_capture == 2
282
+
assert report.capture_health.hours_total == 48
283
+
assert report.capture_health.coverage_ratio is None
284
+
assert report.capture_health.last_segment_at == _utc_ms("20260410", 14, 30)
285
+
assert report.capture_health.facets_with_recent_capture == ("work",)
286
+
assert report.capture_health.facets_silent_24h == ()
287
+
assert report.synthesis_health.activities_count == 3
288
+
assert report.synthesis_health.activities_with_participation == 1
289
+
assert report.synthesis_health.activities_with_story == 1
290
+
assert report.synthesis_health.activities_user_edited == 1
291
+
assert report.synthesis_health.activities_anticipated_unfilled == 1
292
+
assert report.synthesis_health.talent_run_failures_24h == 1
293
+
assert report.synthesis_health.indexer_last_rebuild_at == indexer_ms
294
+
295
+
296
+
def test_activities_with_participation_counts_truthy_field(tmp_path, monkeypatch):
297
+
_configure_env(tmp_path, monkeypatch)
298
+
_set_now(monkeypatch, _utc_dt("20260410"))
299
+
_minimal_facet_tree(tmp_path)
300
+
_write_activity(
301
+
tmp_path,
302
+
"work",
303
+
"20260410",
304
+
_activity_record("20260410", "one", participation=[{"entity_id": "a"}]),
305
+
)
306
+
_write_activity(
307
+
tmp_path,
308
+
"work",
309
+
"20260410",
310
+
_activity_record("20260410", "two", participation=[]),
311
+
)
312
+
_write_activity(
313
+
tmp_path,
314
+
"work",
315
+
"20260410",
316
+
_activity_record("20260410", "three", hidden=True, participation=[{"a": 1}]),
317
+
)
318
+
319
+
report = health_surface.summary("20260410")
320
+
321
+
assert report.synthesis_health.activities_with_participation == 1
322
+
323
+
324
+
def test_hours_total_is_24_times_range_days(tmp_path, monkeypatch):
325
+
_configure_env(tmp_path, monkeypatch)
326
+
_set_now(monkeypatch, _utc_dt("20260410"))
327
+
_minimal_facet_tree(tmp_path)
328
+
329
+
report = health_surface.for_range("20260408", "20260410")
330
+
331
+
assert report.capture_health.hours_total == 72
332
+
333
+
334
+
def test_facets_partition_into_recent_or_silent_24h(tmp_path, monkeypatch):
335
+
_configure_env(tmp_path, monkeypatch)
336
+
_set_now(monkeypatch, _utc_dt("20260410"))
337
+
_minimal_facet_tree(tmp_path, facets=("home", "work"))
338
+
_write_activity(
339
+
tmp_path,
340
+
"work",
341
+
"20260410",
342
+
_activity_record("20260410", "work_100000_1800", segments=["100000_1800"]),
343
+
)
344
+
345
+
report = health_surface.summary("20260410")
346
+
347
+
assert report.facets == ("home", "work")
348
+
assert report.capture_health.facets_with_recent_capture == ("work",)
349
+
assert report.capture_health.facets_silent_24h == ("home",)
350
+
assert sorted(
351
+
report.capture_health.facets_with_recent_capture
352
+
+ report.capture_health.facets_silent_24h
353
+
) == list(report.facets)
354
+
355
+
356
+
def test_profile_entities_total_lives_on_consumer_signal(tmp_path, monkeypatch):
357
+
_configure_env(tmp_path, monkeypatch)
358
+
_set_now(monkeypatch, _utc_dt("20260410"))
359
+
_minimal_facet_tree(tmp_path)
360
+
_write_entity(tmp_path, "alex", "Alex")
361
+
_write_entity(tmp_path, "blair", "Blair")
362
+
363
+
report = health_surface.summary("20260410")
364
+
365
+
assert report.consumer_signal.profile_entities_total == 2
366
+
assert not hasattr(report.synthesis_health, "profile_entities_total")
367
+
368
+
369
+
def test_silent_facet_note_ladder_thresholds(tmp_path, monkeypatch):
370
+
_configure_env(tmp_path, monkeypatch)
371
+
_set_now(monkeypatch, _utc_dt("20260410"))
372
+
_minimal_facet_tree(
373
+
tmp_path,
374
+
facets=("criticalf", "fresh", "infof", "never", "warnf"),
375
+
)
376
+
_write_activity(
377
+
tmp_path,
378
+
"fresh",
379
+
"20260410",
380
+
_activity_record("20260410", "fresh_110000_1800", segments=["110000_1800"]),
381
+
)
382
+
_write_activity(
383
+
tmp_path,
384
+
"infof",
385
+
"20260409",
386
+
_activity_record("20260409", "info_100000_1800", segments=["100000_1800"]),
387
+
)
388
+
_write_activity(
389
+
tmp_path,
390
+
"warnf",
391
+
"20260407",
392
+
_activity_record("20260407", "warn_100000_1800", segments=["100000_1800"]),
393
+
)
394
+
_write_activity(
395
+
tmp_path,
396
+
"criticalf",
397
+
"20260403",
398
+
_activity_record(
399
+
"20260403",
400
+
"critical_100000_1800",
401
+
segments=["100000_1800"],
402
+
),
403
+
)
404
+
405
+
report = health_surface.summary("20260410")
406
+
note_map = {
407
+
note.message.split(":", 1)[0]: note
408
+
for note in report.notes
409
+
if note.category == "capture" and ":" in note.message
410
+
}
411
+
412
+
assert "fresh" not in note_map
413
+
assert note_map["infof"].severity == "info"
414
+
assert "last capture" in note_map["infof"].message
415
+
assert note_map["warnf"].severity == "warn"
416
+
assert note_map["criticalf"].severity == "critical"
417
+
assert note_map["never"].severity == "info"
418
+
assert "no captures recorded" in note_map["never"].message
419
+
420
+
421
+
def test_silent_facet_emits_single_highest_severity_note(tmp_path, monkeypatch):
422
+
_configure_env(tmp_path, monkeypatch)
423
+
_set_now(monkeypatch, _utc_dt("20260410"))
424
+
_minimal_facet_tree(tmp_path, facets=("criticalf",))
425
+
_write_activity(
426
+
tmp_path,
427
+
"criticalf",
428
+
"20260403",
429
+
_activity_record(
430
+
"20260403",
431
+
"critical_100000_1800",
432
+
segments=["100000_1800"],
433
+
),
434
+
)
435
+
436
+
report = health_surface.summary("20260410")
437
+
critical_notes = [
438
+
note
439
+
for note in report.notes
440
+
if note.category == "capture" and note.message.startswith("criticalf:")
441
+
]
442
+
443
+
assert len(critical_notes) == 1
444
+
assert critical_notes[0].severity == "critical"
445
+
446
+
447
+
def test_indexer_stale_warn_threshold(tmp_path, monkeypatch):
448
+
_configure_env(tmp_path, monkeypatch)
449
+
fixed_now = _utc_dt("20260410")
450
+
_set_now(monkeypatch, fixed_now)
451
+
_minimal_facet_tree(tmp_path)
452
+
_write_indexer_db(tmp_path, fixed_now - timedelta(days=8))
453
+
454
+
report = health_surface.summary("20260410")
455
+
456
+
assert any(
457
+
note.category == "synthesis"
458
+
and note.severity == "warn"
459
+
and "indexer database last rebuilt" in note.message
460
+
for note in report.notes
461
+
)
462
+
463
+
464
+
def test_indexer_missing_emits_warn_and_none_timestamp(tmp_path, monkeypatch):
465
+
_configure_env(tmp_path, monkeypatch)
466
+
_set_now(monkeypatch, _utc_dt("20260410"))
467
+
_minimal_facet_tree(tmp_path)
468
+
469
+
report = health_surface.summary("20260410")
470
+
471
+
assert report.synthesis_health.indexer_last_rebuild_at is None
472
+
assert any(
473
+
note.category == "synthesis"
474
+
and note.severity == "warn"
475
+
and "indexer database missing" in note.message
476
+
for note in report.notes
477
+
)
478
+
479
+
480
+
def test_missing_talent_day_indexes_emit_info(tmp_path, monkeypatch):
481
+
_configure_env(tmp_path, monkeypatch)
482
+
_set_now(monkeypatch, _utc_dt("20260410"))
483
+
_minimal_facet_tree(tmp_path)
484
+
485
+
report = health_surface.summary("20260410")
486
+
487
+
assert report.synthesis_health.talent_run_failures_24h is None
488
+
assert any(
489
+
note.category == "synthesis"
490
+
and note.severity == "info"
491
+
and "talent day-index logs missing" in note.message
492
+
for note in report.notes
493
+
)
494
+
495
+
496
+
def test_for_range_defaults_to_last_7_days_ending_today(tmp_path, monkeypatch):
497
+
_configure_env(tmp_path, monkeypatch)
498
+
_set_now(monkeypatch, _utc_dt("20260410"))
499
+
_minimal_facet_tree(tmp_path)
500
+
501
+
report = health_surface.for_range()
502
+
503
+
assert report.range == ("20260404", "20260410")
504
+
505
+
506
+
def test_summary_defaults_to_today_utc(tmp_path, monkeypatch):
507
+
_configure_env(tmp_path, monkeypatch)
508
+
_set_now(monkeypatch, _utc_dt("20260410"))
509
+
_minimal_facet_tree(tmp_path)
510
+
511
+
report = health_surface.summary()
512
+
513
+
assert report.range == ("20260410", "20260410")
514
+
515
+
516
+
def test_for_range_validation_errors(tmp_path, monkeypatch):
517
+
_configure_env(tmp_path, monkeypatch)
518
+
_set_now(monkeypatch, _utc_dt("20260410"))
519
+
_minimal_facet_tree(tmp_path)
520
+
521
+
with pytest.raises(ValueError, match="both endpoints or neither"):
522
+
health_surface.for_range(day_from="20260410")
523
+
with pytest.raises(ValueError, match="day_from must be <="):
524
+
health_surface.for_range("20260411", "20260410")
525
+
with pytest.raises(ValueError, match="day must match YYYYMMDD"):
526
+
health_surface.for_range("2026-04-10", "20260410")
527
+
528
+
529
+
def test_report_notes_sorted_by_severity_category_message(tmp_path, monkeypatch):
530
+
_configure_env(tmp_path, monkeypatch)
531
+
_set_now(monkeypatch, _utc_dt("20260410"))
532
+
_minimal_facet_tree(tmp_path, facets=("alpha",))
533
+
534
+
report = health_surface.summary("20260410")
535
+
ordered = [(note.severity, note.category, note.message) for note in report.notes]
536
+
537
+
assert ordered == [
538
+
(
539
+
"warn",
540
+
"synthesis",
541
+
"indexer database missing at journal/indexer/journal.sqlite; search-backed consumers may be stale.",
542
+
),
543
+
("info", "capture", "alpha: no captures recorded in the last 7 days."),
544
+
(
545
+
"info",
546
+
"capture",
547
+
"coverage_ratio unavailable in v1 — expected-hours denominator arrives Sprint 5+",
548
+
),
549
+
(
550
+
"info",
551
+
"synthesis",
552
+
"corrections roll-up not available — corrections ledger exists only from Sprint 5+",
553
+
),
554
+
(
555
+
"info",
556
+
"synthesis",
557
+
"talent day-index logs missing for 20260410, 20260409; last-24h failure count unavailable.",
558
+
),
559
+
]
560
+
561
+
562
+
def test_consumer_signal_counts_compose_ledger(tmp_path, monkeypatch):
563
+
_configure_env(tmp_path, monkeypatch)
564
+
now = datetime.now(UTC)
565
+
day_recent = now.strftime("%Y%m%d")
566
+
day_stale = (now - timedelta(days=30)).strftime("%Y%m%d")
567
+
_minimal_facet_tree(tmp_path)
568
+
_write_entity(tmp_path, "alex", "Alex")
569
+
_write_activity(
570
+
tmp_path,
571
+
"work",
572
+
day_recent,
573
+
_activity_record(
574
+
day_recent,
575
+
"recent_090000_1800",
576
+
commitments=[
577
+
{
578
+
"owner": "Alex",
579
+
"owner_entity_id": "alex",
580
+
"action": "send recap",
581
+
"counterparty": "Blair",
582
+
"counterparty_entity_id": "blair",
583
+
"context": "Recent open item",
584
+
}
585
+
],
586
+
created_at=int((now - timedelta(days=1)).timestamp() * 1000),
587
+
),
588
+
)
589
+
_write_activity(
590
+
tmp_path,
591
+
"work",
592
+
day_stale,
593
+
_activity_record(
594
+
day_stale,
595
+
"stale_090000_1800",
596
+
commitments=[
597
+
{
598
+
"owner": "Alex",
599
+
"owner_entity_id": "alex",
600
+
"action": "draft proposal",
601
+
"counterparty": "Blair",
602
+
"counterparty_entity_id": "blair",
603
+
"context": "Stale open item",
604
+
}
605
+
],
606
+
created_at=int((now - timedelta(days=30)).timestamp() * 1000),
607
+
),
608
+
)
609
+
610
+
from think.surfaces import ledger as ledger_surface
611
+
612
+
original_list = ledger_surface.list
613
+
calls: list[dict[str, object]] = []
614
+
615
+
def spy_list(**kwargs):
616
+
calls.append(dict(kwargs))
617
+
return original_list(**kwargs)
618
+
619
+
monkeypatch.setattr(health_surface.ledger, "list", spy_list)
620
+
621
+
report = health_surface.summary(day_recent)
622
+
623
+
assert report.consumer_signal.ledger_open_items_total == len(
624
+
original_list(state="open")
625
+
)
626
+
assert report.consumer_signal.ledger_stale_items_count == len(
627
+
original_list(state="open", age_days_gte=14)
628
+
)
629
+
assert calls == [{"state": "open"}, {"state": "open", "age_days_gte": 14}]
630
+
631
+
632
+
def test_structural_info_notes_are_always_present_and_coverage_ratio_is_none(
633
+
tmp_path, monkeypatch
634
+
):
635
+
_configure_env(tmp_path, monkeypatch)
636
+
_set_now(monkeypatch, _utc_dt("20260410"))
637
+
_minimal_facet_tree(tmp_path)
638
+
639
+
report = health_surface.summary("20260410")
640
+
641
+
assert report.capture_health.coverage_ratio is None
642
+
assert all(note.detected_at == report.generated_at for note in report.notes)
643
+
note_tuples = [
644
+
(
645
+
note.severity,
646
+
note.category,
647
+
note.message,
648
+
note.detail_pointer,
649
+
)
650
+
for note in report.notes
651
+
]
652
+
assert (
653
+
"info",
654
+
"capture",
655
+
"coverage_ratio unavailable in v1 — expected-hours denominator arrives Sprint 5+",
656
+
_SPEC_POINTER,
657
+
) in note_tuples
658
+
assert (
659
+
"info",
660
+
"synthesis",
661
+
"corrections roll-up not available — corrections ledger exists only from Sprint 5+",
662
+
_SPEC_POINTER,
663
+
) in note_tuples
664
+
665
+
666
+
def test_segment_crossing_midnight_clipping(tmp_path, monkeypatch):
667
+
_configure_env(tmp_path, monkeypatch)
668
+
_set_now(monkeypatch, _utc_dt("20260410"))
669
+
_minimal_facet_tree(tmp_path)
670
+
_write_activity(
671
+
tmp_path,
672
+
"work",
673
+
"20260410",
674
+
_activity_record(
675
+
"20260410",
676
+
"night_233000_3600",
677
+
segments=["233000_3600"],
678
+
),
679
+
)
680
+
681
+
report = health_surface.summary("20260410")
682
+
683
+
assert report.capture_health.hours_with_capture == 1
684
+
685
+
686
+
def test_activities_anticipated_unfilled_counts_past_visible_only(
687
+
tmp_path, monkeypatch
688
+
):
689
+
_configure_env(tmp_path, monkeypatch)
690
+
_set_now(monkeypatch, _utc_dt("20260410"))
691
+
_minimal_facet_tree(tmp_path)
692
+
_write_activity(
693
+
tmp_path,
694
+
"work",
695
+
"20260410",
696
+
_activity_record(
697
+
"20260410",
698
+
"anticipated_past",
699
+
source="anticipated",
700
+
start=_iso_utc("20260410", 9),
701
+
),
702
+
)
703
+
_write_activity(
704
+
tmp_path,
705
+
"work",
706
+
"20260410",
707
+
_activity_record(
708
+
"20260410",
709
+
"anticipated_hidden",
710
+
source="anticipated",
711
+
start=_iso_utc("20260410", 8),
712
+
hidden=True,
713
+
),
714
+
)
715
+
_write_activity(
716
+
tmp_path,
717
+
"work",
718
+
"20260410",
719
+
_activity_record(
720
+
"20260410",
721
+
"anticipated_future",
722
+
source="anticipated",
723
+
start=_iso_utc("20260410", 18),
724
+
),
725
+
)
726
+
727
+
report = health_surface.summary("20260410")
728
+
729
+
assert report.synthesis_health.activities_anticipated_unfilled == 1
730
+
731
+
732
+
def test_activities_user_edited_counts_prefixed_actors(tmp_path, monkeypatch):
733
+
_configure_env(tmp_path, monkeypatch)
734
+
_set_now(monkeypatch, _utc_dt("20260410"))
735
+
_minimal_facet_tree(tmp_path)
736
+
_write_activity(
737
+
tmp_path,
738
+
"work",
739
+
"20260410",
740
+
_activity_record(
741
+
"20260410",
742
+
"cli_edit",
743
+
edits=[{"actor": "cli:update", "fields": ["details"]}],
744
+
),
745
+
)
746
+
_write_activity(
747
+
tmp_path,
748
+
"work",
749
+
"20260410",
750
+
_activity_record(
751
+
"20260410",
752
+
"owner_edit",
753
+
edits=[{"actor": "owner", "fields": ["details"]}],
754
+
),
755
+
)
756
+
_write_activity(
757
+
tmp_path,
758
+
"work",
759
+
"20260410",
760
+
_activity_record(
761
+
"20260410",
762
+
"user_edit",
763
+
edits=[{"actor": "user", "fields": ["details"]}],
764
+
),
765
+
)
766
+
_write_activity(
767
+
tmp_path,
768
+
"work",
769
+
"20260410",
770
+
_activity_record(
771
+
"20260410",
772
+
"system_edit",
773
+
edits=[{"actor": "system:story", "fields": ["story"]}],
774
+
),
775
+
)
776
+
777
+
report = health_surface.summary("20260410")
778
+
779
+
assert report.synthesis_health.activities_user_edited == 3
780
+
781
+
782
+
def test_talent_run_failures_24h_use_today_and_yesterday_day_indices(
783
+
tmp_path, monkeypatch
784
+
):
785
+
_configure_env(tmp_path, monkeypatch)
786
+
_set_now(monkeypatch, _utc_dt("20260410"))
787
+
_minimal_facet_tree(tmp_path)
788
+
_write_talent_day(
789
+
tmp_path,
790
+
"20260410",
791
+
{
792
+
"use_id": "1",
793
+
"name": "flow",
794
+
"day": "20260410",
795
+
"facet": None,
796
+
"ts": _utc_ms("20260410", 9),
797
+
"status": "completed",
798
+
"error": "boom",
799
+
},
800
+
{
801
+
"use_id": "2",
802
+
"name": "flow",
803
+
"day": "20260410",
804
+
"facet": None,
805
+
"ts": _utc_ms("20260410", 8),
806
+
"status": "completed",
807
+
},
808
+
)
809
+
_write_talent_day(
810
+
tmp_path,
811
+
"20260409",
812
+
{
813
+
"use_id": "3",
814
+
"name": "flow",
815
+
"day": "20260409",
816
+
"facet": None,
817
+
"ts": _utc_ms("20260409", 15),
818
+
"status": "failed",
819
+
},
820
+
{
821
+
"use_id": "4",
822
+
"name": "flow",
823
+
"day": "20260409",
824
+
"facet": None,
825
+
"ts": _utc_ms("20260409", 11, 30),
826
+
"status": "failed",
827
+
},
828
+
)
829
+
830
+
report = health_surface.summary("20260410")
831
+
832
+
assert report.synthesis_health.talent_run_failures_24h == 2
833
+
834
+
835
+
def test_cli_health_summary_full_range_json(tmp_path, monkeypatch):
836
+
_configure_env(tmp_path, monkeypatch)
837
+
_set_now(monkeypatch, _utc_dt("20260410"))
838
+
_minimal_facet_tree(tmp_path)
839
+
840
+
from think.call import call_app
841
+
842
+
summary_result = _RUNNER.invoke(call_app, ["health", "summary", "--json"])
843
+
full_result = _RUNNER.invoke(call_app, ["health", "full", "--json"])
844
+
range_result = _RUNNER.invoke(
845
+
call_app,
846
+
[
847
+
"health",
848
+
"for-range",
849
+
"--day-from",
850
+
"20260409",
851
+
"--day-to",
852
+
"20260410",
853
+
"--json",
854
+
],
855
+
)
856
+
857
+
for result in (summary_result, full_result, range_result):
858
+
assert result.exit_code == 0
859
+
payload = json.loads(result.stdout)
860
+
assert set(payload) == {
861
+
"generated_at",
862
+
"range",
863
+
"facets",
864
+
"capture_health",
865
+
"synthesis_health",
866
+
"consumer_signal",
867
+
"notes",
868
+
}
869
+
870
+
871
+
def test_cli_help_disambiguates_and_lists_health_once(tmp_path, monkeypatch):
872
+
_configure_env(tmp_path, monkeypatch)
873
+
_minimal_facet_tree(tmp_path)
874
+
875
+
from think.call import call_app
876
+
877
+
health_help = _RUNNER.invoke(call_app, ["health", "--help"])
878
+
root_help = _RUNNER.invoke(call_app, ["--help"])
879
+
normalized_help = " ".join(health_help.stdout.split())
880
+
881
+
assert health_help.exit_code == 0
882
+
assert (
883
+
"Health: journal-data trust signals (for infrastructure/service liveness, use `sol health`)."
884
+
in normalized_help
885
+
)
886
+
assert root_help.exit_code == 0
887
+
assert (
888
+
sum(
889
+
1
890
+
for line in root_help.stdout.splitlines()
891
+
if re.search(r"│\s+health\s+Health:", line)
892
+
)
893
+
== 1
894
+
)
895
+
896
+
897
+
def test_cli_pipeline_relocated_behavior(tmp_path, monkeypatch):
898
+
_configure_env(tmp_path, monkeypatch)
899
+
_minimal_facet_tree(tmp_path)
900
+
day = "20260101"
901
+
expected_today = datetime.now().strftime("%Y%m%d")
902
+
expected_yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
903
+
_write_pipeline_log(
904
+
tmp_path,
905
+
day,
906
+
"123_segment.jsonl",
907
+
{"event": "run.start", "mode": "segment"},
908
+
{"event": "talent.dispatch", "mode": "segment"},
909
+
{"event": "talent.complete", "mode": "segment"},
910
+
{"event": "run.complete", "mode": "segment", "duration_ms": 42},
911
+
)
912
+
913
+
from think.call import call_app
914
+
915
+
default_result = _RUNNER.invoke(call_app, ["health", "pipeline"])
916
+
day_result = _RUNNER.invoke(call_app, ["health", "pipeline", "--day", day])
917
+
yesterday_result = _RUNNER.invoke(call_app, ["health", "pipeline", "--yesterday"])
918
+
error_result = _RUNNER.invoke(
919
+
call_app,
920
+
["health", "pipeline", "--day", day, "--yesterday"],
921
+
)
922
+
923
+
assert default_result.exit_code == 0
924
+
assert json.loads(default_result.stdout)["day"] == expected_today
925
+
assert day_result.exit_code == 0
926
+
day_payload = json.loads(day_result.stdout)
927
+
assert day_payload["day"] == day
928
+
assert day_payload["runs"]["segment"]["count"] == 1
929
+
assert day_payload["talents"]["dispatched"] >= 1
930
+
assert yesterday_result.exit_code == 0
931
+
assert json.loads(yesterday_result.stdout)["day"] == expected_yesterday
932
+
assert error_result.exit_code == 1
933
+
assert "mutually exclusive" in error_result.stderr
+3
-1
think/call.py
+3
-1
think/call.py
···
73
73
74
74
# Mount built-in CLIs (not auto-discovered since they live under think/)
75
75
from think.tools.call import app as journal_app
76
+
from think.tools.health import app as health_app
76
77
from think.tools.ledger import app as ledger_app
77
78
from think.tools.navigate import app as navigate_app
78
79
from think.tools.profile import app as profile_app
79
80
from think.tools.routines import app as routines_app
80
81
from think.tools.sol import app as sol_app
81
82
83
+
call_app.add_typer(health_app, name="health")
82
84
call_app.add_typer(journal_app, name="journal")
83
85
call_app.add_typer(ledger_app, name="ledger")
86
+
call_app.add_typer(navigate_app, name="navigate")
84
87
call_app.add_typer(profile_app, name="profile")
85
-
call_app.add_typer(navigate_app, name="navigate")
86
88
call_app.add_typer(routines_app, name="routines")
87
89
call_app.add_typer(sol_app, name="identity")
88
90
+518
think/surfaces/health.py
+518
think/surfaces/health.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
"""Health consumer surface for journal-data trust signals.
5
+
6
+
This surface reports on capture, synthesis, and consumer-facing trust signals
7
+
derived from journal data. For infrastructure and service liveness, use
8
+
``sol health`` instead.
9
+
"""
10
+
11
+
from __future__ import annotations
12
+
13
+
import json
14
+
from dataclasses import dataclass
15
+
from datetime import UTC, datetime, timedelta
16
+
from pathlib import Path
17
+
from typing import Any, Iterator
18
+
19
+
from think.activities import load_activity_records
20
+
from think.entities.journal import load_all_journal_entities
21
+
from think.facets import get_facets
22
+
from think.surfaces import ledger
23
+
from think.surfaces.types import (
24
+
CaptureHealth,
25
+
ConsumerSignalHealth,
26
+
HealthNote,
27
+
HealthReport,
28
+
SynthesisHealth,
29
+
)
30
+
from think.utils import get_journal, segment_parse
31
+
32
+
FACET_SILENT_INFO_HOURS = 24
33
+
# 24h is the first trust-signal rung: a facet that has been quiet for a full day should surface as informational drift before it becomes an operational concern.
34
+
FACET_SILENT_WARN_HOURS = 72
35
+
# 72h is a stronger silence signal than 24h without yet implying an outright break; warn keeps the ladder graduated before the weekly threshold.
36
+
FACET_SILENT_CRITICAL_HOURS = 168
37
+
# 168h (7d) is the highest silent-facet rung: a facet quiet for a full week is likely broken, muted in practice, or missing expected capture.
38
+
INDEXER_STALE_WARN_DAYS = 7
39
+
# 7d matches the weekly freshness bar for search-backed consumers; shorter windows would over-warn on journals that intentionally rebuild less often.
40
+
LEDGER_STALE_DAYS = 14
41
+
# 14d mirrors the consumer-signal stale-item threshold so the health surface stays aligned with ledger backlog review.
42
+
USER_EDIT_ACTOR_PREFIXES = ("cli:", "owner", "user")
43
+
# These prefixes identify operator- or user-authored corrections without trying to enumerate every internal automation actor string.
44
+
_DAY_MS = 86_400_000
45
+
_HOUR_MS = 3_600_000
46
+
_SPEC_POINTER = "cpo/specs/in-flight/consumer-surface-health.md"
47
+
48
+
49
+
def _resolve_now() -> datetime:
50
+
return datetime.now(UTC)
51
+
52
+
53
+
@dataclass(frozen=True)
54
+
class _ScanAggregate:
55
+
capture_hour_slots: frozenset[tuple[str, int]]
56
+
last_segment_at: int | None
57
+
activities_count: int
58
+
activities_with_participation: int
59
+
activities_with_story: int
60
+
activities_user_edited: int
61
+
activities_anticipated_unfilled: int
62
+
63
+
64
+
def _resolve_day(day: str) -> str:
65
+
try:
66
+
datetime.strptime(day, "%Y%m%d")
67
+
except ValueError as exc:
68
+
raise ValueError("day must match YYYYMMDD") from exc
69
+
return day
70
+
71
+
72
+
def _resolve_day_start(day: str) -> datetime:
73
+
return datetime.strptime(_resolve_day(day), "%Y%m%d").replace(tzinfo=UTC)
74
+
75
+
76
+
def _parse_segment_bounds(
77
+
raw_segment: str, day: str
78
+
) -> tuple[datetime, datetime] | None:
79
+
start_time, _ = segment_parse(raw_segment)
80
+
if start_time is None:
81
+
return None
82
+
83
+
try:
84
+
duration_seconds = int(raw_segment.split("_", 1)[1])
85
+
except (IndexError, ValueError):
86
+
return None
87
+
88
+
start_of_day = _resolve_day_start(day)
89
+
start_dt = start_of_day.replace(
90
+
hour=start_time.hour,
91
+
minute=start_time.minute,
92
+
second=start_time.second,
93
+
microsecond=0,
94
+
)
95
+
return start_dt, start_dt + timedelta(seconds=duration_seconds)
96
+
97
+
98
+
def _iter_range_facet_days(day_from: str, day_to: str) -> Iterator[tuple[str, str]]:
99
+
start_day = _resolve_day_start(day_from)
100
+
end_day = _resolve_day_start(day_to)
101
+
facets = tuple(sorted(get_facets().keys()))
102
+
current = start_day
103
+
while current <= end_day:
104
+
day = current.strftime("%Y%m%d")
105
+
for facet in facets:
106
+
yield facet, day
107
+
current += timedelta(days=1)
108
+
109
+
110
+
def _parse_segment_hour_slots(segments: object, day: str) -> set[tuple[str, int]]:
111
+
if not isinstance(segments, list):
112
+
return set()
113
+
114
+
slots: set[tuple[str, int]] = set()
115
+
start_of_day = _resolve_day_start(day)
116
+
start_of_next_day = start_of_day + timedelta(days=1)
117
+
for raw_segment in segments:
118
+
if not isinstance(raw_segment, str):
119
+
continue
120
+
bounds = _parse_segment_bounds(raw_segment, day)
121
+
if bounds is None:
122
+
continue
123
+
start_dt, end_dt = bounds
124
+
clipped_end = min(end_dt, start_of_next_day)
125
+
current = start_dt.replace(minute=0, second=0, microsecond=0)
126
+
while current < clipped_end:
127
+
slots.add((day, current.hour))
128
+
current += timedelta(hours=1)
129
+
return slots
130
+
131
+
132
+
def _count_user_edits(record: dict[str, Any]) -> int:
133
+
edits = record.get("edits")
134
+
if not isinstance(edits, list):
135
+
return 0
136
+
137
+
count = 0
138
+
for edit in edits:
139
+
if not isinstance(edit, dict):
140
+
continue
141
+
actor = edit.get("actor")
142
+
if isinstance(actor, str) and actor.startswith(USER_EDIT_ACTOR_PREFIXES):
143
+
count += 1
144
+
return count
145
+
146
+
147
+
def _scan_records(day_from: str, day_to: str) -> _ScanAggregate:
148
+
capture_hour_slots: set[tuple[str, int]] = set()
149
+
last_segment_at: int | None = None
150
+
activities_count = 0
151
+
activities_with_participation = 0
152
+
activities_with_story = 0
153
+
activities_user_edited = 0
154
+
activities_anticipated_unfilled = 0
155
+
generated_at_ms = int(_resolve_now().timestamp() * 1000)
156
+
157
+
for facet, day in _iter_range_facet_days(day_from, day_to):
158
+
for record in load_activity_records(facet, day, include_hidden=True):
159
+
capture_hour_slots.update(
160
+
_parse_segment_hour_slots(record.get("segments"), day)
161
+
)
162
+
163
+
segments = record.get("segments")
164
+
if isinstance(segments, list):
165
+
for raw_segment in segments:
166
+
if not isinstance(raw_segment, str):
167
+
continue
168
+
bounds = _parse_segment_bounds(raw_segment, day)
169
+
if bounds is None:
170
+
continue
171
+
_, end_dt = bounds
172
+
end_ms = int(end_dt.timestamp() * 1000)
173
+
if last_segment_at is None or end_ms > last_segment_at:
174
+
last_segment_at = end_ms
175
+
176
+
if bool(record.get("hidden", False)):
177
+
continue
178
+
179
+
activities_count += 1
180
+
if record.get("participation"):
181
+
activities_with_participation += 1
182
+
if record.get("story"):
183
+
activities_with_story += 1
184
+
if _count_user_edits(record) > 0:
185
+
activities_user_edited += 1
186
+
187
+
if record.get("source") == "anticipated" and not bool(
188
+
record.get("cancelled", False)
189
+
):
190
+
start_value = record.get("start")
191
+
if isinstance(start_value, str):
192
+
try:
193
+
start_dt = datetime.fromisoformat(
194
+
start_value.replace("Z", "+00:00")
195
+
)
196
+
except ValueError:
197
+
start_dt = None
198
+
if start_dt is not None:
199
+
if start_dt.tzinfo is None:
200
+
start_dt = start_dt.replace(tzinfo=UTC)
201
+
if int(start_dt.timestamp() * 1000) <= generated_at_ms:
202
+
activities_anticipated_unfilled += 1
203
+
204
+
return _ScanAggregate(
205
+
capture_hour_slots=frozenset(capture_hour_slots),
206
+
last_segment_at=last_segment_at,
207
+
activities_count=activities_count,
208
+
activities_with_participation=activities_with_participation,
209
+
activities_with_story=activities_with_story,
210
+
activities_user_edited=activities_user_edited,
211
+
activities_anticipated_unfilled=activities_anticipated_unfilled,
212
+
)
213
+
214
+
215
+
def _last_segment_ts_per_facet() -> dict[str, int | None]:
216
+
facets = tuple(sorted(get_facets().keys()))
217
+
current = _resolve_now()
218
+
days = [
219
+
(current - timedelta(days=offset)).strftime("%Y%m%d")
220
+
for offset in range(7, -1, -1)
221
+
]
222
+
last_seen: dict[str, int | None] = {facet: None for facet in facets}
223
+
224
+
for facet in facets:
225
+
for day in days:
226
+
for record in load_activity_records(facet, day, include_hidden=True):
227
+
segments = record.get("segments")
228
+
if not isinstance(segments, list):
229
+
continue
230
+
for raw_segment in segments:
231
+
if not isinstance(raw_segment, str):
232
+
continue
233
+
bounds = _parse_segment_bounds(raw_segment, day)
234
+
if bounds is None:
235
+
continue
236
+
_, end_dt = bounds
237
+
end_ms = int(end_dt.timestamp() * 1000)
238
+
if last_seen[facet] is None or end_ms > last_seen[facet]:
239
+
last_seen[facet] = end_ms
240
+
241
+
return last_seen
242
+
243
+
244
+
def _build_capture_health(
245
+
aggregate: _ScanAggregate,
246
+
report_range: tuple[str, str],
247
+
facets: tuple[str, ...],
248
+
generated_at: int,
249
+
) -> tuple[CaptureHealth, list[HealthNote]]:
250
+
last_seen_by_facet = _last_segment_ts_per_facet()
251
+
recent_cutoff = generated_at - (FACET_SILENT_INFO_HOURS * _HOUR_MS)
252
+
recent_facets = tuple(
253
+
sorted(
254
+
facet
255
+
for facet in facets
256
+
if last_seen_by_facet.get(facet) is not None
257
+
and int(last_seen_by_facet[facet] or 0) >= recent_cutoff
258
+
)
259
+
)
260
+
recent_facet_set = set(recent_facets)
261
+
silent_facets = tuple(
262
+
sorted(facet for facet in facets if facet not in recent_facet_set)
263
+
)
264
+
265
+
day_from, day_to = report_range
266
+
hours_total = (
267
+
(_resolve_day_start(day_to) - _resolve_day_start(day_from)).days + 1
268
+
) * 24
269
+
notes = [
270
+
HealthNote(
271
+
severity="info",
272
+
category="capture",
273
+
message="coverage_ratio unavailable in v1 — expected-hours denominator arrives Sprint 5+",
274
+
detected_at=generated_at,
275
+
detail_pointer=_SPEC_POINTER,
276
+
)
277
+
]
278
+
279
+
for facet in facets:
280
+
last_seen = last_seen_by_facet.get(facet)
281
+
if last_seen is None:
282
+
notes.append(
283
+
HealthNote(
284
+
severity="info",
285
+
category="capture",
286
+
message=f"{facet}: no captures recorded in the last 7 days.",
287
+
detected_at=generated_at,
288
+
detail_pointer=None,
289
+
)
290
+
)
291
+
continue
292
+
293
+
gap_hours = max(0, (generated_at - last_seen) // _HOUR_MS)
294
+
if gap_hours >= FACET_SILENT_CRITICAL_HOURS:
295
+
severity = "critical"
296
+
elif gap_hours >= FACET_SILENT_WARN_HOURS:
297
+
severity = "warn"
298
+
elif gap_hours >= FACET_SILENT_INFO_HOURS:
299
+
severity = "info"
300
+
else:
301
+
continue
302
+
303
+
last_seen_text = (
304
+
datetime.fromtimestamp(last_seen / 1000, tz=UTC)
305
+
.isoformat()
306
+
.replace("+00:00", "Z")
307
+
)
308
+
notes.append(
309
+
HealthNote(
310
+
severity=severity,
311
+
category="capture",
312
+
message=f"{facet}: last capture {gap_hours}h ago ({last_seen_text}).",
313
+
detected_at=generated_at,
314
+
detail_pointer=None,
315
+
)
316
+
)
317
+
318
+
return (
319
+
CaptureHealth(
320
+
hours_with_capture=len(aggregate.capture_hour_slots),
321
+
hours_total=hours_total,
322
+
coverage_ratio=None,
323
+
facets_with_recent_capture=recent_facets,
324
+
facets_silent_24h=silent_facets,
325
+
last_segment_at=aggregate.last_segment_at,
326
+
),
327
+
notes,
328
+
)
329
+
330
+
331
+
def _build_synthesis_health(
332
+
aggregate: _ScanAggregate,
333
+
generated_at: int,
334
+
) -> tuple[SynthesisHealth, list[HealthNote]]:
335
+
notes = [
336
+
HealthNote(
337
+
severity="info",
338
+
category="synthesis",
339
+
message="corrections roll-up not available — corrections ledger exists only from Sprint 5+",
340
+
detected_at=generated_at,
341
+
detail_pointer=_SPEC_POINTER,
342
+
)
343
+
]
344
+
345
+
generated_at_dt = datetime.fromtimestamp(generated_at / 1000, tz=UTC)
346
+
talent_days = (
347
+
generated_at_dt.strftime("%Y%m%d"),
348
+
(generated_at_dt - timedelta(days=1)).strftime("%Y%m%d"),
349
+
)
350
+
talents_dir = Path(get_journal()) / "talents"
351
+
talent_rows: list[dict[str, Any]] = []
352
+
missing_talent_days: list[str] = []
353
+
for day in talent_days:
354
+
path = talents_dir / f"{day}.jsonl"
355
+
if not path.exists():
356
+
missing_talent_days.append(day)
357
+
continue
358
+
with path.open(encoding="utf-8") as handle:
359
+
for raw_line in handle:
360
+
line = raw_line.strip()
361
+
if not line:
362
+
continue
363
+
try:
364
+
payload = json.loads(line)
365
+
except ValueError:
366
+
continue
367
+
if isinstance(payload, dict):
368
+
talent_rows.append(payload)
369
+
370
+
talent_run_failures_24h: int | None
371
+
if missing_talent_days:
372
+
talent_run_failures_24h = None
373
+
notes.append(
374
+
HealthNote(
375
+
severity="info",
376
+
category="synthesis",
377
+
message=(
378
+
"talent day-index logs missing for "
379
+
+ ", ".join(missing_talent_days)
380
+
+ "; last-24h failure count unavailable."
381
+
),
382
+
detected_at=generated_at,
383
+
detail_pointer=None,
384
+
)
385
+
)
386
+
else:
387
+
talent_run_failures_24h = 0
388
+
cutoff = generated_at - _DAY_MS
389
+
for row in talent_rows:
390
+
try:
391
+
timestamp = int(row.get("ts")) # type: ignore[arg-type]
392
+
except (TypeError, ValueError):
393
+
continue
394
+
if timestamp < cutoff or timestamp > generated_at:
395
+
continue
396
+
status = row.get("status")
397
+
if row.get("error") or status not in ("ok", "completed", None):
398
+
talent_run_failures_24h += 1
399
+
400
+
indexer_path = Path(get_journal()) / "indexer" / "journal.sqlite"
401
+
if not indexer_path.exists():
402
+
indexer_last_rebuild_at = None
403
+
notes.append(
404
+
HealthNote(
405
+
severity="warn",
406
+
category="synthesis",
407
+
message="indexer database missing at journal/indexer/journal.sqlite; search-backed consumers may be stale.",
408
+
detected_at=generated_at,
409
+
detail_pointer=None,
410
+
)
411
+
)
412
+
else:
413
+
indexer_last_rebuild_at = indexer_path.stat().st_mtime_ns // 1_000_000
414
+
if generated_at - indexer_last_rebuild_at > INDEXER_STALE_WARN_DAYS * _DAY_MS:
415
+
stale_days = (generated_at - indexer_last_rebuild_at) // _DAY_MS
416
+
notes.append(
417
+
HealthNote(
418
+
severity="warn",
419
+
category="synthesis",
420
+
message=(
421
+
f"indexer database last rebuilt {stale_days}d ago; "
422
+
"search-backed consumers may be stale."
423
+
),
424
+
detected_at=generated_at,
425
+
detail_pointer=None,
426
+
)
427
+
)
428
+
429
+
return (
430
+
SynthesisHealth(
431
+
activities_count=aggregate.activities_count,
432
+
activities_with_participation=aggregate.activities_with_participation,
433
+
activities_with_story=aggregate.activities_with_story,
434
+
activities_user_edited=aggregate.activities_user_edited,
435
+
activities_anticipated_unfilled=aggregate.activities_anticipated_unfilled,
436
+
talent_run_failures_24h=talent_run_failures_24h,
437
+
indexer_last_rebuild_at=indexer_last_rebuild_at,
438
+
),
439
+
notes,
440
+
)
441
+
442
+
443
+
def _build_consumer_signal_health() -> ConsumerSignalHealth:
444
+
return ConsumerSignalHealth(
445
+
ledger_open_items_total=len(ledger.list(state="open")),
446
+
ledger_stale_items_count=len(
447
+
ledger.list(state="open", age_days_gte=LEDGER_STALE_DAYS)
448
+
),
449
+
profile_entities_total=len(load_all_journal_entities()),
450
+
)
451
+
452
+
453
+
def _build_report(day_from: str, day_to: str) -> HealthReport:
454
+
generated_at = int(_resolve_now().timestamp() * 1000)
455
+
facets = tuple(sorted(get_facets().keys()))
456
+
aggregate = _scan_records(day_from, day_to)
457
+
capture_health, capture_notes = _build_capture_health(
458
+
aggregate,
459
+
(day_from, day_to),
460
+
facets,
461
+
generated_at,
462
+
)
463
+
synthesis_health, synthesis_notes = _build_synthesis_health(
464
+
aggregate,
465
+
generated_at,
466
+
)
467
+
notes = capture_notes + synthesis_notes
468
+
severity_rank = {"critical": 0, "warn": 1, "info": 2}
469
+
notes.sort(
470
+
key=lambda note: (
471
+
severity_rank.get(note.severity, 99),
472
+
note.category,
473
+
note.message,
474
+
)
475
+
)
476
+
return HealthReport(
477
+
generated_at=generated_at,
478
+
range=(day_from, day_to),
479
+
facets=facets,
480
+
capture_health=capture_health,
481
+
synthesis_health=synthesis_health,
482
+
consumer_signal=_build_consumer_signal_health(),
483
+
notes=tuple(notes),
484
+
)
485
+
486
+
487
+
def summary(day: str | None = None) -> HealthReport:
488
+
resolved_day = (
489
+
_resolve_now().strftime("%Y%m%d") if day is None else _resolve_day(day)
490
+
)
491
+
return _build_report(resolved_day, resolved_day)
492
+
493
+
494
+
def full(day: str | None = None) -> HealthReport:
495
+
resolved_day = (
496
+
_resolve_now().strftime("%Y%m%d") if day is None else _resolve_day(day)
497
+
)
498
+
return _build_report(resolved_day, resolved_day)
499
+
500
+
501
+
def for_range(
502
+
day_from: str | None = None,
503
+
day_to: str | None = None,
504
+
) -> HealthReport:
505
+
if day_from is None and day_to is None:
506
+
today = _resolve_now()
507
+
day_to = today.strftime("%Y%m%d")
508
+
day_from = (today - timedelta(days=6)).strftime("%Y%m%d")
509
+
elif day_from is None or day_to is None:
510
+
raise ValueError("both endpoints or neither")
511
+
else:
512
+
_resolve_day(day_from)
513
+
_resolve_day(day_to)
514
+
515
+
if day_from > day_to:
516
+
raise ValueError("day_from must be <= day_to")
517
+
518
+
return _build_report(day_from, day_to)
+48
think/surfaces/types.py
+48
think/surfaces/types.py
···
77
77
decisions_involving_them: tuple[Decision, ...]
78
78
sources: tuple[ActivitySourceRef, ...]
79
79
generated_at: int
80
+
81
+
82
+
@dataclass(frozen=True)
83
+
class CaptureHealth:
84
+
hours_with_capture: int
85
+
hours_total: int
86
+
coverage_ratio: float | None
87
+
facets_with_recent_capture: tuple[str, ...]
88
+
facets_silent_24h: tuple[str, ...]
89
+
last_segment_at: int | None
90
+
91
+
92
+
@dataclass(frozen=True)
93
+
class SynthesisHealth:
94
+
activities_count: int
95
+
activities_with_participation: int
96
+
activities_with_story: int
97
+
activities_user_edited: int
98
+
activities_anticipated_unfilled: int
99
+
talent_run_failures_24h: int | None
100
+
indexer_last_rebuild_at: int | None
101
+
102
+
103
+
@dataclass(frozen=True)
104
+
class ConsumerSignalHealth:
105
+
ledger_open_items_total: int
106
+
ledger_stale_items_count: int
107
+
profile_entities_total: int
108
+
109
+
110
+
@dataclass(frozen=True)
111
+
class HealthNote:
112
+
severity: str
113
+
category: str
114
+
message: str
115
+
detected_at: int
116
+
detail_pointer: str | None
117
+
118
+
119
+
@dataclass(frozen=True)
120
+
class HealthReport:
121
+
generated_at: int
122
+
range: tuple[str, str]
123
+
facets: tuple[str, ...]
124
+
capture_health: CaptureHealth
125
+
synthesis_health: SynthesisHealth
126
+
consumer_signal: ConsumerSignalHealth
127
+
notes: tuple[HealthNote, ...]
+182
think/tools/health.py
+182
think/tools/health.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
from __future__ import annotations
5
+
6
+
import dataclasses
7
+
import json
8
+
import json as jsonlib
9
+
from datetime import datetime, timedelta
10
+
from typing import Optional
11
+
12
+
import typer
13
+
14
+
from think.pipeline_health import summarize_pipeline_day
15
+
from think.surfaces import health as health_surface
16
+
from think.surfaces.types import HealthReport
17
+
from think.utils import require_solstone
18
+
19
+
app = typer.Typer(
20
+
help="Health: journal-data trust signals (for infrastructure/service liveness, use `sol health`).",
21
+
no_args_is_help=True,
22
+
)
23
+
24
+
25
+
@app.callback()
26
+
def callback() -> None:
27
+
require_solstone()
28
+
29
+
30
+
def _echo_json(payload: object) -> None:
31
+
typer.echo(jsonlib.dumps(payload, indent=2, sort_keys=False))
32
+
33
+
34
+
def _render_summary(report: HealthReport) -> None:
35
+
typer.echo(f"Range: {report.range[0]} -> {report.range[1]}")
36
+
typer.echo("Capture")
37
+
typer.echo(f" hours_with_capture: {report.capture_health.hours_with_capture}")
38
+
typer.echo(f" hours_total: {report.capture_health.hours_total}")
39
+
typer.echo(f" coverage_ratio: {report.capture_health.coverage_ratio}")
40
+
typer.echo(
41
+
" facets_with_recent_capture: "
42
+
+ ", ".join(report.capture_health.facets_with_recent_capture)
43
+
)
44
+
typer.echo(
45
+
" facets_silent_24h: " + ", ".join(report.capture_health.facets_silent_24h)
46
+
)
47
+
typer.echo(f" last_segment_at: {report.capture_health.last_segment_at}")
48
+
typer.echo("Synthesis")
49
+
typer.echo(f" activities_count: {report.synthesis_health.activities_count}")
50
+
typer.echo(
51
+
" activities_with_participation: "
52
+
+ str(report.synthesis_health.activities_with_participation)
53
+
)
54
+
typer.echo(
55
+
f" activities_with_story: {report.synthesis_health.activities_with_story}"
56
+
)
57
+
typer.echo(
58
+
f" activities_user_edited: {report.synthesis_health.activities_user_edited}"
59
+
)
60
+
typer.echo(
61
+
" activities_anticipated_unfilled: "
62
+
+ str(report.synthesis_health.activities_anticipated_unfilled)
63
+
)
64
+
typer.echo(
65
+
" talent_run_failures_24h: "
66
+
+ str(report.synthesis_health.talent_run_failures_24h)
67
+
)
68
+
typer.echo(
69
+
" indexer_last_rebuild_at: "
70
+
+ str(report.synthesis_health.indexer_last_rebuild_at)
71
+
)
72
+
typer.echo("Consumer Signals")
73
+
typer.echo(
74
+
f" ledger_open_items_total: {report.consumer_signal.ledger_open_items_total}"
75
+
)
76
+
typer.echo(
77
+
f" ledger_stale_items_count: {report.consumer_signal.ledger_stale_items_count}"
78
+
)
79
+
typer.echo(
80
+
f" profile_entities_total: {report.consumer_signal.profile_entities_total}"
81
+
)
82
+
typer.echo("Notes")
83
+
if not report.notes:
84
+
typer.echo(" none")
85
+
return
86
+
for note in report.notes:
87
+
typer.echo(f" [{note.severity}] {note.category}: {note.message}")
88
+
89
+
90
+
def _render_full(report: HealthReport) -> None:
91
+
_render_summary(report)
92
+
if not report.capture_health.facets_silent_24h:
93
+
return
94
+
95
+
typer.echo("Silent Facet Detail")
96
+
for facet in report.capture_health.facets_silent_24h:
97
+
matching = [
98
+
note
99
+
for note in report.notes
100
+
if note.category == "capture" and note.message.startswith(f"{facet}:")
101
+
]
102
+
if not matching:
103
+
continue
104
+
for note in matching:
105
+
typer.echo(f" {facet}: [{note.severity}] {note.message}")
106
+
107
+
108
+
@app.command("summary")
109
+
def summary(
110
+
day: str | None = typer.Option(None, "--day"),
111
+
json: bool = typer.Option(False, "--json"),
112
+
) -> None:
113
+
"""Summarize journal-data trust signals for one day."""
114
+
try:
115
+
report = health_surface.summary(day=day)
116
+
except ValueError as exc:
117
+
raise typer.BadParameter(str(exc)) from exc
118
+
if json:
119
+
_echo_json(dataclasses.asdict(report))
120
+
return
121
+
_render_summary(report)
122
+
123
+
124
+
@app.command("full")
125
+
def full(
126
+
day: str | None = typer.Option(None, "--day"),
127
+
json: bool = typer.Option(False, "--json"),
128
+
) -> None:
129
+
"""Render the full journal-data trust report for one day."""
130
+
try:
131
+
report = health_surface.full(day=day)
132
+
except ValueError as exc:
133
+
raise typer.BadParameter(str(exc)) from exc
134
+
if json:
135
+
_echo_json(dataclasses.asdict(report))
136
+
return
137
+
_render_full(report)
138
+
139
+
140
+
@app.command("for-range")
141
+
def for_range(
142
+
day_from: str | None = typer.Option(None, "--day-from"),
143
+
day_to: str | None = typer.Option(None, "--day-to"),
144
+
json: bool = typer.Option(False, "--json"),
145
+
) -> None:
146
+
"""Render the journal-data trust report for an inclusive day range."""
147
+
try:
148
+
report = health_surface.for_range(day_from=day_from, day_to=day_to)
149
+
except ValueError as exc:
150
+
raise typer.BadParameter(str(exc)) from exc
151
+
if json:
152
+
_echo_json(dataclasses.asdict(report))
153
+
return
154
+
_render_full(report)
155
+
156
+
157
+
@app.command(
158
+
"pipeline",
159
+
help="Thin wrapper around think.pipeline_health. For journal-data trust checks use `summary` / `full` / `for-range`.",
160
+
)
161
+
def pipeline(
162
+
day: Optional[str] = typer.Option(
163
+
None, "--day", help="Day to summarize (YYYYMMDD)."
164
+
),
165
+
yesterday: bool = typer.Option(
166
+
False, "--yesterday", help="Summarize yesterday's pipeline."
167
+
),
168
+
) -> None:
169
+
"""Summarize think pipeline health for one day."""
170
+
if day is not None and yesterday:
171
+
typer.echo("--day and --yesterday are mutually exclusive", err=True)
172
+
raise typer.Exit(1)
173
+
174
+
if day is not None:
175
+
target = day
176
+
elif yesterday:
177
+
target = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
178
+
else:
179
+
target = datetime.now().strftime("%Y%m%d")
180
+
181
+
summary = summarize_pipeline_day(target)
182
+
typer.echo(json.dumps(summary, indent=2, sort_keys=False))