personal memory agent
feat(think): --no-activity-prompts flag for cheap realizer backfill
Adds an opt-in `--no-activity-prompts` flag to `sol think` that lets
`--segments` / `--segment` runs write realized activity records but skip
the per-activity `run_activity_prompts(...)` cogitate calls that drive
backfill cost (today: just `todos:todo`).
Default behavior is byte-identical. Flag is incompatible with the
standalone `--activity` mode (errors at parse time). Each skip emits an
`activity.prompts_skipped` JSONL breadcrumb so backfill runs are
auditable.
+269
+230
tests/test_think_no_activity_prompts.py
+230
tests/test_think_no_activity_prompts.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
"""Tests for think --no-activity-prompts behavior."""
5
+
6
+
import json
7
+
from pathlib import Path
8
+
9
+
import pytest
10
+
11
+
DAY = "20240115"
12
+
SEGMENT = "120000_300"
13
+
STREAM = "default"
14
+
FACET = "work"
15
+
ACTIVITY_ID = "coding_120000_300"
16
+
17
+
18
+
@pytest.fixture
19
+
def segment_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
20
+
"""Create a temporary journal with a segment directory."""
21
+
journal = tmp_path / "journal"
22
+
segment_path = journal / "chronicle" / DAY / STREAM / SEGMENT
23
+
(segment_path / "talents").mkdir(parents=True)
24
+
25
+
monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
26
+
return segment_path
27
+
28
+
29
+
def _segment_configs(*names: str) -> dict[str, dict]:
30
+
configs = {
31
+
"sense": {
32
+
"priority": 10,
33
+
"type": "generate",
34
+
"output": "json",
35
+
"schedule": "segment",
36
+
},
37
+
}
38
+
return {name: dict(configs[name]) for name in names}
39
+
40
+
41
+
def _write_sense_output(segment_dir: Path, sense_json: dict) -> None:
42
+
(segment_dir / "talents" / "sense.json").write_text(
43
+
json.dumps(sense_json),
44
+
encoding="utf-8",
45
+
)
46
+
47
+
48
+
class EndedActivityStateMachine:
49
+
def __init__(self, journal_root: Path) -> None:
50
+
self.state: dict = {}
51
+
self.last_segment_key: str | None = None
52
+
self.last_segment_day: str | None = None
53
+
self.journal_root = journal_root
54
+
55
+
def update(self, sense_output: dict, segment: str, day: str) -> list[dict]:
56
+
self.last_segment_key = segment
57
+
self.last_segment_day = day
58
+
self.state = {
59
+
FACET: {
60
+
"facet": FACET,
61
+
"state": "active",
62
+
"id": ACTIVITY_ID,
63
+
}
64
+
}
65
+
return [{"state": "ended", "id": ACTIVITY_ID, "facet": FACET}]
66
+
67
+
def get_completed_activities(self) -> list[dict]:
68
+
return [
69
+
{
70
+
"id": ACTIVITY_ID,
71
+
"activity": "coding",
72
+
"segments": [SEGMENT],
73
+
"level_avg": 0.5,
74
+
"description": "coding",
75
+
"active_entities": [],
76
+
"created_at": 1713200000000,
77
+
}
78
+
]
79
+
80
+
81
+
def _patch_segment_dependencies(
82
+
monkeypatch: pytest.MonkeyPatch,
83
+
append_calls: list[tuple],
84
+
activity_calls: list[dict],
85
+
) -> None:
86
+
from think import thinking as think
87
+
88
+
monkeypatch.setattr(
89
+
think,
90
+
"get_talent_configs",
91
+
lambda schedule=None, **kwargs: _segment_configs("sense"),
92
+
)
93
+
monkeypatch.setattr(
94
+
think,
95
+
"cortex_request",
96
+
lambda prompt, name, config=None: f"agent-{name}",
97
+
)
98
+
monkeypatch.setattr(
99
+
think,
100
+
"wait_for_uses",
101
+
lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
102
+
)
103
+
monkeypatch.setattr(
104
+
think,
105
+
"append_activity_record",
106
+
lambda *args: append_calls.append(args),
107
+
)
108
+
monkeypatch.setattr(
109
+
think,
110
+
"run_activity_prompts",
111
+
lambda **kwargs: activity_calls.append(kwargs) or True,
112
+
)
113
+
monkeypatch.setattr(think, "_callosum", None)
114
+
115
+
116
+
def test_flag_set_skips_prompts_but_writes_record(
117
+
segment_dir: Path,
118
+
monkeypatch: pytest.MonkeyPatch,
119
+
) -> None:
120
+
from think import thinking as think
121
+
from think.thinking import ThinkingJSONLWriter
122
+
123
+
append_calls: list[tuple] = []
124
+
activity_calls: list[dict] = []
125
+
jsonl_path = segment_dir.parent.parent / "health" / "test_no_prompts.jsonl"
126
+
writer = ThinkingJSONLWriter(str(jsonl_path))
127
+
128
+
_write_sense_output(
129
+
segment_dir,
130
+
{"density": "active", "recommend": {}, "facets": []},
131
+
)
132
+
_patch_segment_dependencies(monkeypatch, append_calls, activity_calls)
133
+
monkeypatch.setattr(think, "_jsonl", writer)
134
+
135
+
think.run_segment_sense(
136
+
DAY,
137
+
SEGMENT,
138
+
refresh=False,
139
+
verbose=False,
140
+
stream=STREAM,
141
+
state_machine=EndedActivityStateMachine(segment_dir.parents[3]),
142
+
skip_activity_prompts=True,
143
+
)
144
+
writer.close()
145
+
146
+
assert len(append_calls) >= 1
147
+
assert activity_calls == []
148
+
149
+
events = [
150
+
json.loads(line)
151
+
for line in jsonl_path.read_text(encoding="utf-8").splitlines()
152
+
if line.strip()
153
+
]
154
+
assert any(
155
+
event["event"] == "activity.prompts_skipped"
156
+
and event["activity"] == ACTIVITY_ID
157
+
and event["facet"] == FACET
158
+
for event in events
159
+
)
160
+
161
+
162
+
def test_flag_unset_runs_prompts_unchanged(
163
+
segment_dir: Path,
164
+
monkeypatch: pytest.MonkeyPatch,
165
+
) -> None:
166
+
from think import thinking as think
167
+
168
+
append_calls: list[tuple] = []
169
+
activity_calls: list[dict] = []
170
+
171
+
_write_sense_output(
172
+
segment_dir,
173
+
{"density": "active", "recommend": {}, "facets": []},
174
+
)
175
+
_patch_segment_dependencies(monkeypatch, append_calls, activity_calls)
176
+
177
+
think.run_segment_sense(
178
+
DAY,
179
+
SEGMENT,
180
+
refresh=False,
181
+
verbose=False,
182
+
stream=STREAM,
183
+
state_machine=EndedActivityStateMachine(segment_dir.parents[3]),
184
+
)
185
+
186
+
assert len(append_calls) >= 1
187
+
assert activity_calls == [
188
+
{
189
+
"day": DAY,
190
+
"activity_id": ACTIVITY_ID,
191
+
"facet": FACET,
192
+
"refresh": False,
193
+
"verbose": False,
194
+
"max_concurrency": 2,
195
+
}
196
+
]
197
+
198
+
199
+
def test_flag_with_activity_mode_errors(
200
+
tmp_path: Path,
201
+
monkeypatch: pytest.MonkeyPatch,
202
+
capsys: pytest.CaptureFixture[str],
203
+
) -> None:
204
+
from think import thinking as think
205
+
206
+
journal = tmp_path / "journal"
207
+
(journal / "chronicle" / DAY).mkdir(parents=True)
208
+
monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
209
+
monkeypatch.setenv("SOL_SKIP_SUPERVISOR_CHECK", "1")
210
+
monkeypatch.setattr(
211
+
"sys.argv",
212
+
[
213
+
"sol think",
214
+
"--activity",
215
+
ACTIVITY_ID,
216
+
"--facet",
217
+
FACET,
218
+
"--day",
219
+
DAY,
220
+
"--no-activity-prompts",
221
+
],
222
+
)
223
+
224
+
with pytest.raises(SystemExit) as excinfo:
225
+
think.main()
226
+
227
+
assert excinfo.value.code == 2
228
+
stderr = capsys.readouterr().err
229
+
assert "--no-activity-prompts" in stderr
230
+
assert "--activity" in stderr
+39
think/thinking.py
+39
think/thinking.py
···
468
468
stream: str | None = None,
469
469
timeout: int | None = 610,
470
470
state_machine: ActivityStateMachine | None = None,
471
+
*,
472
+
skip_activity_prompts: bool = False,
471
473
) -> tuple[int, int, list[str]]:
472
474
"""Run Sense-first linear orchestrator for a single segment.
473
475
···
704
706
activity_id,
705
707
facet,
706
708
)
709
+
if skip_activity_prompts:
710
+
_jsonl_log(
711
+
"activity.prompts_skipped",
712
+
day=day,
713
+
segment=segment,
714
+
activity=str(activity_id),
715
+
facet=str(facet),
716
+
mode=target_schedule,
717
+
reason="--no-activity-prompts",
718
+
)
719
+
continue
707
720
run_activity_prompts(
708
721
day=routing_day,
709
722
activity_id=str(activity_id),
···
968
981
activity_id,
969
982
facet,
970
983
)
984
+
if skip_activity_prompts:
985
+
_jsonl_log(
986
+
"activity.prompts_skipped",
987
+
day=day,
988
+
segment=segment,
989
+
activity=str(activity_id),
990
+
facet=str(facet),
991
+
mode=target_schedule,
992
+
reason="--no-activity-prompts",
993
+
)
994
+
continue
971
995
run_activity_prompts(
972
996
day=routing_day,
973
997
activity_id=str(activity_id),
···
2801
2825
help="Disable per-batch agent wait timeout in --segments mode",
2802
2826
)
2803
2827
parser.add_argument(
2828
+
"--no-activity-prompts",
2829
+
action="store_true",
2830
+
help=(
2831
+
"Write realized activity records but skip per-activity cogitate runs "
2832
+
'(schedule="activity" talents). Used by realizer backfill to write '
2833
+
"activity records cheaply without firing per-activity prompts. "
2834
+
"Incompatible with --activity."
2835
+
),
2836
+
)
2837
+
parser.add_argument(
2804
2838
"--updated",
2805
2839
action="store_true",
2806
2840
help="List days with pending daily processing and exit",
···
2879
2913
if args.activity and not args.day:
2880
2914
parser.error("--activity requires --day")
2881
2915
2916
+
if args.no_activity_prompts and args.activity:
2917
+
parser.error("--no-activity-prompts cannot be combined with --activity")
2918
+
2882
2919
if args.activity and (args.segment or args.segments or args.flush):
2883
2920
parser.error(
2884
2921
"--activity is incompatible with --segment, --segments, and --flush"
···
3004
3041
stream=seg_stream,
3005
3042
timeout=None if args.no_timeout else 610,
3006
3043
state_machine=batch_state_machine,
3044
+
skip_activity_prompts=args.no_activity_prompts,
3007
3045
)
3008
3046
# Touch stream.updated marker after each segment
3009
3047
try:
···
3117
3155
stream=resolved_stream,
3118
3156
timeout=None if args.no_timeout else 610,
3119
3157
state_machine=ActivityStateMachine(journal_root=Path(get_journal())),
3158
+
skip_activity_prompts=args.no_activity_prompts,
3120
3159
)
3121
3160
else:
3122
3161
success_count, fail_count, failed_names = run_daily_prompts(