personal memory agent
1#!/usr/bin/env python3
2# SPDX-License-Identifier: AGPL-3.0-only
3# Copyright (c) 2026 sol pbc
4
5"""Cogitate prompt command-span lint.
6
7This is the prompt-side static mirror of ``solstone/think/cogitate_policy.py``.
8It lints command-bearing code spans in every ``type == "cogitate"`` talent
9prompt, applying the same command classification to prompt markdown that the
10runtime policy applies to live tool calls.
11
12The committed ``ALLOWLIST`` is keyed by ``(file, kind)`` with an allowed count.
13This is a ``!=`` self-check: a live count above the allowlisted count fails as a
14new/over violation, and a live count below the allowlisted count fails as a
15stale entry. The gate iterates the union of discovered keys and allowlist keys
16so stale entries for vanished files are still visited.
17
18No live ``sol`` or ``journal`` command is invoked at lint time.
19
20Known limitation: single-backtick inline spans are parsed; double-backtick spans
21are not.
22
23Exit codes:
24 0 - clean
25 1 - any over violation or stale allowlist entry
26"""
27
28from __future__ import annotations
29
30import argparse
31import os
32import re
33import shlex
34import sys
35from pathlib import Path
36
37import frontmatter
38
39import solstone.think.talent as talent
40from solstone.think.cogitate_contract import (
41 COGITATE_JOURNAL_COMMANDS,
42 cogitate_journal_command_list,
43)
44
45ROOT = Path(__file__).resolve().parent.parent
46
47# Must equal solstone/think/cogitate_policy.py:_READ_TOOLS
48# (cogitate_policy.py:24). Duplicated here intentionally; no shared import.
49READ_TOOLS = frozenset({"read_file", "glob", "list_directory", "grep_search"})
50
51# Claude Code tool names the cogitate runtime lacks. Case-sensitive: lowercase
52# glob/grep_search are sanctioned read tools and must pass. Runtime tool names
53# write_file/replace are intentionally absent; prompts may reference them in
54# negative instructions.
55CLI_AGENT_TOOLS = frozenset({"Read", "Edit", "Write", "Bash", "Glob", "Grep", "Agent"})
56
57SHELL_READ_COMMANDS = frozenset({"cat", "ls", "head", "tail", "less", "more"})
58SHELL_OPERATOR_CHARS = frozenset("();<>|&")
59SHELL_WRAPPERS = frozenset(
60 {
61 "bash",
62 "sh",
63 "zsh",
64 "dash",
65 "env",
66 "eval",
67 "exec",
68 "command",
69 "xargs",
70 "sudo",
71 "python",
72 "python3",
73 }
74)
75
76_FENCED_BLOCK_RE = re.compile(r"```[^\n]*\n(.*?)```", re.S)
77_INLINE_CODE_RE = re.compile(r"`([^`\n]+)`")
78_EMBEDDED_SOL_JOURNAL_RE = re.compile(r"(?<![\w./-])(?:sol|journal)(?![\w/-])")
79_PLACEHOLDER_RE = re.compile(r"<[+A-Za-z0-9_][+A-Za-z0-9_. -]*>")
80
81# Curated, minimal: lowest-confidence class, NOT anchored in cogitate_policy.py.
82# A false entry (flagging a flag the CLI accepts) is worse than omission.
83# NOTE: 'facets --json' was a scope seed but is VALID
84# (sol call facets list-candidates accepts --json, facets/call.py:40) -- omitted.
85UNSUPPORTED_FLAGS: list[tuple[tuple[str, ...], str, str]] = []
86
87ALLOWLIST: dict[tuple[str, str], int] = {}
88
89JOURNAL_ALTERNATIVE = (
90 "use `journal` with one of {"
91 + cogitate_journal_command_list()
92 + "}, or use `sol`/`sol call`"
93)
94READ_ALTERNATIVE = (
95 "use a bounded read tool: read_file, list_directory, glob, or grep_search"
96)
97PROMPT_CONTRACT = (
98 "cogitate prompts must name only on-contract command forms: "
99 "`sol`/`sol call`, approved `journal` subcommands, or bounded read tools"
100)
101
102
103def command_tokens(command: str) -> list[str]:
104 """Return shell-like tokens, falling back to whitespace splitting."""
105 try:
106 return shlex.split(command)
107 except ValueError:
108 return command.split()
109
110
111def _shell_syntax_violation(command: str) -> bool:
112 """Return True when command uses shell syntax outside quoted data."""
113 if "$(" in command or "`" in command:
114 return True
115 if "\n" in command or "\r" in command:
116 return True
117 index = 0
118 length = len(command)
119 quote: str | None = None
120 while index < length:
121 char = command[index]
122 if quote == "'":
123 if char == "'":
124 quote = None
125 elif quote == '"':
126 if char == "\\" and index + 1 < length:
127 index += 2
128 continue
129 if char == '"':
130 quote = None
131 else:
132 if char == "\\" and index + 1 < length:
133 index += 2
134 continue
135 if char in ("'", '"'):
136 quote = char
137 elif char in SHELL_OPERATOR_CHARS:
138 return True
139 index += 1
140 # Diverges from the runtime scanner: prompt lint receives per-line spans from
141 # fenced examples, so a legitimate multi-line quoted value can leave the
142 # first line with an open quote. That is documentation, not shell composition.
143 return False
144
145
146def _wrapper_embeds_sol_or_journal(command: str, tokens: list[str]) -> bool:
147 if not tokens or tokens[0] not in SHELL_WRAPPERS:
148 return False
149 return _EMBEDDED_SOL_JOURNAL_RE.search(command) is not None
150
151
152def _raw_substitution_or_multiline(command: str) -> bool:
153 return "$(" in command or "`" in command or "\n" in command or "\r" in command
154
155
156def _mask_placeholders(command: str) -> str:
157 return _PLACEHOLDER_RE.sub("ARG", command)
158
159
160def _shell_composition_finding(command: str, tokens: list[str]) -> bool:
161 if _raw_substitution_or_multiline(command):
162 return True
163 if _wrapper_embeds_sol_or_journal(command, tokens):
164 return True
165 if tokens[0] in {"sol", "journal"} and _shell_syntax_violation(
166 _mask_placeholders(command)
167 ):
168 return True
169 return False
170
171
172def extract_command_spans(body: str) -> list[tuple[int, str]]:
173 """Return ``(lineno, span_text)`` for fenced lines and inline code spans."""
174 spans: list[tuple[int, int, str]] = []
175
176 for match in _FENCED_BLOCK_RE.finditer(body):
177 block_start_line = body[: match.start(1)].count("\n") + 1
178 for offset, line in enumerate(match.group(1).splitlines()):
179 stripped = line.strip()
180 if stripped:
181 spans.append(
182 (block_start_line + offset, match.start(1) + offset, stripped)
183 )
184
185 def blank_fence(match: re.Match[str]) -> str:
186 return "\n" * match.group(0).count("\n")
187
188 inline_body = _FENCED_BLOCK_RE.sub(blank_fence, body)
189 for match in _INLINE_CODE_RE.finditer(inline_body):
190 spans.append(
191 (
192 inline_body[: match.start()].count("\n") + 1,
193 match.start(),
194 match.group(1).strip(),
195 )
196 )
197
198 return [(lineno, span) for lineno, _order, span in sorted(spans)]
199
200
201def _is_contiguous_subsequence(tokens: list[str], sequence: tuple[str, ...]) -> bool:
202 if len(sequence) > len(tokens):
203 return False
204 return any(
205 tuple(tokens[index : index + len(sequence)]) == sequence
206 for index in range(len(tokens) - len(sequence) + 1)
207 )
208
209
210def _unsupported_flag_findings(tokens: list[str]) -> list[tuple[str, str]]:
211 findings: list[tuple[str, str]] = []
212 for verb_sequence, flag, alternative in UNSUPPORTED_FLAGS:
213 if flag in tokens and _is_contiguous_subsequence(tokens, verb_sequence):
214 findings.append(
215 (
216 "unsupported-flag",
217 f"unsupported flag {flag!r} after {' '.join(verb_sequence)}; "
218 f"{alternative}",
219 )
220 )
221 return findings
222
223
224def classify_span(command: str) -> list[tuple[str, str]]:
225 """Return ``(kind, detail)`` violations for one command-bearing span."""
226 tokens = command_tokens(command)
227 if not tokens:
228 return []
229
230 findings = _unsupported_flag_findings(tokens)
231
232 if _shell_composition_finding(command, tokens):
233 findings.append(
234 (
235 "shell-composition",
236 "forbidden shell composition; use one `sol` or approved `journal` "
237 "command without pipes, redirects, chaining, substitution, or wrappers",
238 )
239 )
240 return findings
241
242 if tokens[0] == "sol":
243 return findings
244
245 if tokens[0] in READ_TOOLS:
246 return findings
247
248 if (
249 tokens[0] == "journal"
250 and len(tokens) >= 2
251 and tokens[1] not in COGITATE_JOURNAL_COMMANDS
252 ):
253 findings.append(
254 (
255 "bare-journal",
256 f"forbidden `journal {tokens[1]}`; {JOURNAL_ALTERNATIVE}",
257 )
258 )
259
260 for token in tokens:
261 for name in CLI_AGENT_TOOLS:
262 if token == name or token.startswith(f"{name}("):
263 findings.append(
264 (
265 "cli-agent-tool",
266 f"forbidden Claude Code tool `{name}`; {READ_ALTERNATIVE}",
267 )
268 )
269
270 if tokens[0] in SHELL_READ_COMMANDS:
271 findings.append(
272 (
273 "shell-read",
274 f"forbidden shell reader `{tokens[0]}`; {READ_ALTERNATIVE}",
275 )
276 )
277
278 if tokens[0].startswith("journal/"):
279 findings.append(
280 (
281 "raw-journal-path",
282 f"forbidden raw journal path `{tokens[0]}`; {READ_ALTERNATIVE}",
283 )
284 )
285
286 return findings
287
288
289def lint_prompt(body: str) -> list[tuple[int, str, str]]:
290 """Return sorted ``(lineno, kind, detail)`` violations for prompt body."""
291 findings: list[tuple[int, str, str]] = []
292 for lineno, span in extract_command_spans(body):
293 for kind, detail in classify_span(span):
294 findings.append((lineno, kind, detail))
295 findings.sort()
296 return findings
297
298
299def discover_prompts() -> list[tuple[str, str]]:
300 """Return ``(repo-relative-posix-path, body)`` for cogitate prompts."""
301 prompts: list[tuple[str, str]] = []
302 for info in talent.get_talent_configs(type="cogitate").values():
303 path = Path(str(info["path"]))
304 rel = Path(os.path.relpath(path, ROOT)).as_posix()
305 body = frontmatter.load(path).content
306 prompts.append((rel, body))
307 return sorted(prompts)
308
309
310def count_violations() -> dict[tuple[str, str], int]:
311 """Map ``(posix-relpath, kind)`` -> occurrence count across prompts."""
312 counts: dict[tuple[str, str], int] = {}
313 for rel, body in discover_prompts():
314 for _lineno, kind, _detail in lint_prompt(body):
315 key = (rel, kind)
316 counts[key] = counts.get(key, 0) + 1
317 return counts
318
319
320def evaluate(
321 allowlist: dict[tuple[str, str], int],
322) -> tuple[list[str], list[str], list[str]]:
323 """Return ``(over, stale, tracked)`` human-readable lines."""
324 live: dict[tuple[str, str], list[tuple[int, str]]] = {}
325 for rel, body in discover_prompts():
326 for lineno, kind, detail in lint_prompt(body):
327 live.setdefault((rel, kind), []).append((lineno, detail))
328
329 over: list[str] = []
330 stale: list[str] = []
331 tracked: list[str] = []
332
333 for rel, kind in sorted(set(live) | set(allowlist)):
334 key = (rel, kind)
335 findings = sorted(live.get(key, []))
336 count = len(findings)
337 allowed = allowlist.get(key, 0)
338 if count > allowed:
339 lines = ", ".join(str(lineno) for lineno, _detail in findings)
340 details = "; ".join(sorted({detail for _lineno, detail in findings}))
341 over.append(
342 f"{rel}: {kind} count {count} exceeds allowed {allowed} "
343 f"at line(s) {lines}: {details} - {PROMPT_CONTRACT}; see "
344 "the check_cogitate_prompts gate."
345 )
346 elif count < allowed:
347 delete_hint = " (delete it)" if count == 0 else ""
348 stale.append(
349 f"{rel}: {kind} allowlisted at {allowed} but {count} live - "
350 f"lower the entry to {count}{delete_hint} - "
351 "check_cogitate_prompts ratchets toward empty; a cleaned prompt "
352 "removes its entry."
353 )
354 elif allowed:
355 tracked.append(f"{rel}: {count}/{allowed} {kind} (allowlisted)")
356
357 return over, stale, tracked
358
359
360def main(argv: list[str] | None = None) -> int:
361 parser = argparse.ArgumentParser(description="cogitate prompt command-span lint")
362 parser.parse_args(argv)
363
364 over, stale, tracked = evaluate(ALLOWLIST)
365
366 if tracked:
367 print("cogitate-prompts: known violations (allowlisted):")
368 for line in tracked:
369 print(f" {line}")
370 print()
371
372 if over or stale:
373 if over:
374 print("cogitate-prompts: NEW violations:", file=sys.stderr)
375 for line in over:
376 print(f" {line}", file=sys.stderr)
377 print(file=sys.stderr)
378 if stale:
379 print("cogitate-prompts: STALE allowlist entries:", file=sys.stderr)
380 for line in stale:
381 print(f" {line}", file=sys.stderr)
382 print(file=sys.stderr)
383 print(
384 f"{PROMPT_CONTRACT}; lower stale allowlist counts as prompts are cleaned.",
385 file=sys.stderr,
386 )
387 return 1
388
389 print("cogitate-prompts: pass")
390 return 0
391
392
393if __name__ == "__main__":
394 raise SystemExit(main())