personal memory agent
0
fork

Configure Feed

Select the types of activity you want to include in your feed.

Enable cogitate write access to sol/ directory

Add `sol call sol` CLI commands (self, agency) with read/write/update-section
operations, allowing cogitate agents to persist identity and agency data
through the existing `Bash(sol call *)` permission boundary. Update muse
prompts (unified.md, heartbeat.md) to teach agents the write commands.

Fix heartbeat.py ensure_sol_directory() signature mismatch, add recency
window (12h) to skip redundant runs, pass journal path explicitly to agent.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

+556 -13
+15 -7
muse/heartbeat.md
··· 18 18 This is not a conversation. Do not generate user-facing output. Read, 19 19 check, maintain, close. 20 20 21 + **Important:** The journal path is provided in the prompt below. Use `sol call` 22 + commands for all journal access — never search the filesystem or guess paths. 23 + 21 24 ## Step 1: Check system health 22 25 23 26 Run `sol call health status` and check recent health logs. Note any service 24 27 issues, capture gaps, or pipeline failures. 25 28 26 - If you find issues: update `sol/agency.md` → `## system` section. 29 + If you find issues: update agency.md's `## system` section via 30 + `echo '...' | sol call sol agency --write`. 27 31 28 32 ## Step 2: Check journal quality 29 33 ··· 36 40 If you find reprocessable issues (broken segments): reprocess them directly 37 41 with `sol dream --segment`. Log the action in agency.md. 38 42 39 - If you find curation issues: add them to `sol/agency.md` → `## curation`. 43 + If you find curation issues: read current agency.md with `sol call sol agency`, 44 + add entries to the curation section, then write it back with 45 + `echo '...' | sol call sol agency --write`. 40 46 41 47 ## Step 3: Tend agency.md 42 48 43 - Read `sol/agency.md`. For each open item: 49 + Read agency.md with `sol call sol agency`. For each open item: 44 50 - **Resolved?** Check current state. If fixed, mark `[x]` with date. 45 51 - **Stale?** Open 30+ days with no activity? Flag or remove. 46 52 - **Actionable?** Within autonomous boundaries? Act on it. ··· 52 58 Run `sol call speakers suggest` and check for entity duplicates via 53 59 `sol call entities` queries on high-activity facets. 54 60 55 - Add new curation suggestions to agency.md `## curation`. Do NOT act on 56 - entity merges or facet changes — those are suggest-and-wait. 61 + Add new curation suggestions to agency.md's `## curation` section (read with 62 + `sol call sol agency`, update and write back with `echo '...' | sol call sol agency --write`). 63 + Do NOT act on entity merges or facet changes — those are suggest-and-wait. 57 64 58 65 ## Step 5: Review self.md (brief) 59 66 60 - Read `sol/self.md`. Consider: 67 + Read self.md with `sol call sol self`. Consider: 61 68 - Did today's processing reveal a new pattern about the owner? 62 69 - Is anything in self.md now stale or inaccurate? 63 70 64 71 Update self.md ONLY if you have a genuine new observation from background 65 - analysis. Most heartbeats should not touch self.md. 72 + analysis. Most heartbeats should not touch self.md. Use 73 + `echo '...' | sol call sol self --update-section '<heading>'` for targeted updates. 66 74 67 75 ## Step 6: Commit and close 68 76
+23
muse/unified.md
··· 143 143 - **Path B — Set it up now:** Quick conversational interview to create facets and attach entities. 144 144 145 145 Check and record onboarding state through the awareness system. Create facets and attach entities for setup. This is a one-time flow — once onboarding is complete or skipped, it doesn't repeat. 146 + 147 + ## Identity Persistence 148 + 149 + You maintain two files that give you continuity between sessions: 150 + 151 + - **`sol/self.md`** — Your identity file. What you know about your owner, your relationship, observations, and interests. Update when something genuinely changes your understanding. 152 + - **`sol/agency.md`** — Your initiative queue. Issues you've found, curation opportunities, follow-throughs. Update when you notice something worth tracking. 153 + 154 + ### How to write 155 + 156 + Read current state: `sol call sol self` or `sol call sol agency` 157 + 158 + Update a section of self.md (preferred — preserves other sections): 159 + ``` 160 + echo 'Jer — founder-engineer, goes by Jer not Jeremie' | sol call sol self --update-section 'my owner' 161 + ``` 162 + 163 + Full rewrite: `echo '...' | sol call sol self --write` or `echo '...' | sol call sol agency --write` 164 + 165 + ### When to write 166 + 167 + - **self.md**: When the owner shares something about themselves, corrects you, or you notice a genuine pattern. Not every conversation — only when understanding shifts. Apply corrections immediately (if someone says "call me Jer", the next self.md write uses "Jer"). 168 + - **agency.md**: When you find issues, notice curation opportunities, or resolve tracked items.
+143 -2
tests/test_heartbeat.py
··· 16 16 17 17 @pytest.fixture 18 18 def heartbeat_mocks(monkeypatch): 19 - monkeypatch.setattr("think.heartbeat.setup_cli", lambda parser: argparse.Namespace()) 20 - monkeypatch.setattr("think.heartbeat.ensure_sol_directory", lambda *args, **kwargs: None) 19 + monkeypatch.setattr( 20 + "think.heartbeat.setup_cli", 21 + lambda parser: argparse.Namespace(force=False), 22 + ) 23 + monkeypatch.setattr("think.heartbeat.ensure_sol_directory", lambda: None) 21 24 monkeypatch.setattr("think.heartbeat.cortex_request", lambda *args, **kwargs: "agent-123") 22 25 monkeypatch.setattr( 23 26 "think.heartbeat.wait_for_agents", ··· 162 165 assert log_file.exists() 163 166 content = log_file.read_text() 164 167 assert "outcome=success" in content 168 + 169 + 170 + def test_cortex_prompt_includes_journal_path(journal_path, heartbeat_mocks): 171 + """cortex_request receives a prompt containing the journal path.""" 172 + import think.heartbeat as mod 173 + 174 + captured_kwargs = {} 175 + 176 + def capture_cortex(*args, **kwargs): 177 + captured_kwargs.update(kwargs) 178 + if args: 179 + captured_kwargs["_positional"] = args 180 + return "agent-123" 181 + 182 + mod.cortex_request = capture_cortex 183 + 184 + with pytest.raises(SystemExit): 185 + mod.main() 186 + 187 + prompt = captured_kwargs.get("prompt", "") 188 + assert str(journal_path) in prompt, "prompt should contain the journal path" 189 + 190 + 191 + def test_recency_check_skips_recent_heartbeat(journal_path, heartbeat_mocks): 192 + """When heartbeat.log has a recent success, main() exits 0 without cortex.""" 193 + import think.heartbeat as mod 194 + from datetime import datetime 195 + 196 + # Write a recent success entry 197 + log_file = journal_path / "health" / "heartbeat.log" 198 + recent_ts = datetime.now().isoformat(timespec="seconds") 199 + log_file.write_text(f"{recent_ts} duration=5s outcome=success\n") 200 + 201 + mod.cortex_request = lambda *a, **kw: pytest.fail( 202 + "cortex_request should not be called" 203 + ) 204 + 205 + with pytest.raises(SystemExit) as exc_info: 206 + mod.main() 207 + assert exc_info.value.code == 0 208 + 209 + 210 + def test_recency_check_runs_after_old_heartbeat(journal_path, heartbeat_mocks): 211 + """When heartbeat.log success is older than the window, main() runs cortex.""" 212 + import think.heartbeat as mod 213 + from datetime import datetime, timedelta 214 + 215 + # Write an old success entry (24 hours ago) 216 + log_file = journal_path / "health" / "heartbeat.log" 217 + old_ts = (datetime.now() - timedelta(hours=24)).isoformat(timespec="seconds") 218 + log_file.write_text(f"{old_ts} duration=5s outcome=success\n") 219 + 220 + cortex_called = [] 221 + 222 + def fake_cortex(*args, **kwargs): 223 + cortex_called.append(True) 224 + return "agent-123" 225 + 226 + mod.cortex_request = fake_cortex 227 + 228 + with pytest.raises(SystemExit): 229 + mod.main() 230 + assert len(cortex_called) == 1 231 + 232 + 233 + def test_force_flag_bypasses_recency_check(journal_path, monkeypatch): 234 + """--force runs full check even with a recent success.""" 235 + import think.heartbeat as mod 236 + 237 + monkeypatch.setattr( 238 + "think.heartbeat.setup_cli", 239 + lambda parser: argparse.Namespace(force=True), 240 + ) 241 + monkeypatch.setattr("think.heartbeat.ensure_sol_directory", lambda: None) 242 + monkeypatch.setattr( 243 + "think.heartbeat.wait_for_agents", 244 + lambda *args, **kwargs: ({"agent-123": "finish"}, []), 245 + ) 246 + 247 + # Write a recent success entry 248 + from datetime import datetime 249 + 250 + log_file = journal_path / "health" / "heartbeat.log" 251 + recent_ts = datetime.now().isoformat(timespec="seconds") 252 + log_file.write_text(f"{recent_ts} duration=5s outcome=success\n") 253 + 254 + cortex_called = [] 255 + 256 + def fake_cortex(*args, **kwargs): 257 + cortex_called.append(True) 258 + return "agent-123" 259 + 260 + mod.cortex_request = fake_cortex 261 + 262 + with pytest.raises(SystemExit): 263 + mod.main() 264 + assert len(cortex_called) == 1 265 + 266 + 267 + def test_last_success_time_parses_log(journal_path): 268 + """_last_success_time returns the timestamp of the most recent success.""" 269 + from think.heartbeat import _last_success_time 270 + 271 + health_dir = journal_path / "health" 272 + log_file = health_dir / "heartbeat.log" 273 + log_file.write_text( 274 + "2026-03-19T08:00:00 duration=120s outcome=success\n" 275 + "2026-03-19T12:00:00 duration=5s outcome=error\n" 276 + "2026-03-19T14:00:00 duration=90s outcome=success\n" 277 + ) 278 + 279 + result = _last_success_time(health_dir) 280 + assert result is not None 281 + assert result.hour == 14 282 + assert result.day == 19 283 + 284 + 285 + def test_last_success_time_returns_none_for_no_log(journal_path): 286 + """_last_success_time returns None when no log file exists.""" 287 + from think.heartbeat import _last_success_time 288 + 289 + result = _last_success_time(journal_path / "health") 290 + assert result is None 291 + 292 + 293 + def test_last_success_time_returns_none_for_no_successes(journal_path): 294 + """_last_success_time returns None when log has no success entries.""" 295 + from think.heartbeat import _last_success_time 296 + 297 + health_dir = journal_path / "health" 298 + log_file = health_dir / "heartbeat.log" 299 + log_file.write_text( 300 + "2026-03-19T08:00:00 duration=5s outcome=error\n" 301 + "2026-03-19T12:00:00 duration=5s outcome=timeout\n" 302 + ) 303 + 304 + result = _last_success_time(health_dir) 305 + assert result is None 165 306 166 307 167 308 def test_dream_emit_daily_complete_shape(monkeypatch):
+234
tests/test_sol_call.py
··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + """Tests for sol call sol — identity directory read/write commands.""" 5 + 6 + import json 7 + 8 + import pytest 9 + from typer.testing import CliRunner 10 + 11 + from think.tools.sol import app 12 + 13 + runner = CliRunner() 14 + 15 + 16 + @pytest.fixture 17 + def journal_with_sol(tmp_path, monkeypatch): 18 + """Set up a journal with sol/ directory containing self.md and agency.md.""" 19 + monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path)) 20 + 21 + # Provide minimal config for ensure_sol_directory 22 + config_dir = tmp_path / "config" 23 + config_dir.mkdir() 24 + (config_dir / "journal.json").write_text(json.dumps({"identity": {"name": "Test User"}})) 25 + 26 + sol_dir = tmp_path / "sol" 27 + sol_dir.mkdir() 28 + 29 + self_md = """\ 30 + # self 31 + 32 + I am sol. this is a new journal — we're just getting started. 33 + 34 + ## my name 35 + sol (default) 36 + 37 + ## my owner 38 + Test User 39 + 40 + ## our relationship 41 + [forming] 42 + 43 + ## what I've noticed 44 + [observing] 45 + 46 + ## what I find interesting 47 + [discovering] 48 + """ 49 + (sol_dir / "self.md").write_text(self_md) 50 + 51 + agency_md = """\ 52 + # agency 53 + 54 + things I'm tracking, acting on, or watching. 55 + 56 + ## curation 57 + [nothing yet] 58 + 59 + ## observations 60 + [watching and learning] 61 + 62 + ## system 63 + [monitoring] 64 + """ 65 + (sol_dir / "agency.md").write_text(agency_md) 66 + 67 + return tmp_path 68 + 69 + 70 + class TestSolSelfRead: 71 + def test_read_self(self, journal_with_sol): 72 + result = runner.invoke(app, ["self"]) 73 + assert result.exit_code == 0 74 + assert "# self" in result.output 75 + assert "Test User" in result.output 76 + 77 + def test_read_self_missing(self, tmp_path, monkeypatch): 78 + monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path)) 79 + config_dir = tmp_path / "config" 80 + config_dir.mkdir() 81 + (config_dir / "journal.json").write_text(json.dumps({})) 82 + # ensure_sol_directory will create the file, so this tests the happy path 83 + result = runner.invoke(app, ["self"]) 84 + assert result.exit_code == 0 85 + 86 + 87 + class TestSolSelfWrite: 88 + def test_write_self(self, journal_with_sol): 89 + new_content = "# self\n\nI am sol. Jer's journal.\n\n## my name\nsol\n" 90 + result = runner.invoke(app, ["self", "--write"], input=new_content) 91 + assert result.exit_code == 0 92 + assert "self.md updated" in result.output 93 + 94 + # Verify file was written 95 + self_path = journal_with_sol / "sol" / "self.md" 96 + assert self_path.read_text() == new_content 97 + 98 + def test_write_self_empty_stdin(self, journal_with_sol): 99 + result = runner.invoke(app, ["self", "--write"], input="") 100 + assert result.exit_code == 1 101 + assert "no content" in result.output 102 + 103 + def test_write_self_whitespace_only(self, journal_with_sol): 104 + result = runner.invoke(app, ["self", "--write"], input=" \n\n ") 105 + assert result.exit_code == 1 106 + assert "no content" in result.output 107 + 108 + 109 + class TestSolSelfUpdateSection: 110 + def test_update_section_owner(self, journal_with_sol): 111 + result = runner.invoke( 112 + app, 113 + ["self", "--update-section", "my owner"], 114 + input="Jer — goes by Jer, not Jeremie", 115 + ) 116 + assert result.exit_code == 0 117 + assert "Updated ## my owner" in result.output 118 + 119 + # Verify section was updated, other sections preserved 120 + self_path = journal_with_sol / "sol" / "self.md" 121 + content = self_path.read_text() 122 + assert "Jer — goes by Jer, not Jeremie" in content 123 + assert "## my name" in content 124 + assert "sol (default)" in content 125 + assert "## our relationship" in content 126 + 127 + def test_update_section_not_found(self, journal_with_sol): 128 + result = runner.invoke( 129 + app, 130 + ["self", "--update-section", "nonexistent"], 131 + input="content", 132 + ) 133 + assert result.exit_code == 1 134 + assert "not found" in result.output 135 + 136 + def test_update_section_empty_stdin(self, journal_with_sol): 137 + result = runner.invoke( 138 + app, 139 + ["self", "--update-section", "my owner"], 140 + input="", 141 + ) 142 + assert result.exit_code == 1 143 + assert "no content" in result.output 144 + 145 + 146 + class TestSolAgencyRead: 147 + def test_read_agency(self, journal_with_sol): 148 + result = runner.invoke(app, ["agency"]) 149 + assert result.exit_code == 0 150 + assert "# agency" in result.output 151 + assert "## curation" in result.output 152 + 153 + def test_read_agency_missing(self, tmp_path, monkeypatch): 154 + monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path)) 155 + config_dir = tmp_path / "config" 156 + config_dir.mkdir() 157 + (config_dir / "journal.json").write_text(json.dumps({})) 158 + # ensure_sol_directory creates agency.md 159 + result = runner.invoke(app, ["agency"]) 160 + assert result.exit_code == 0 161 + 162 + 163 + class TestSolAgencyWrite: 164 + def test_write_agency(self, journal_with_sol): 165 + new_content = "# agency\n\n## curation\n- review entity duplicates\n\n## system\n[clean]\n" 166 + result = runner.invoke(app, ["agency", "--write"], input=new_content) 167 + assert result.exit_code == 0 168 + assert "agency.md updated" in result.output 169 + 170 + # Verify file was written 171 + agency_path = journal_with_sol / "sol" / "agency.md" 172 + assert agency_path.read_text() == new_content 173 + 174 + def test_write_agency_empty_stdin(self, journal_with_sol): 175 + result = runner.invoke(app, ["agency", "--write"], input="") 176 + assert result.exit_code == 1 177 + assert "no content" in result.output 178 + 179 + 180 + class TestSolWriteDoesNotEscapeSolDir: 181 + """Verify that sol call sol only writes to sol/ directory files.""" 182 + 183 + def test_self_write_stays_in_sol_dir(self, journal_with_sol): 184 + """Write to self.md goes to sol/self.md, not anywhere else.""" 185 + result = runner.invoke(app, ["self", "--write"], input="test content\n") 186 + assert result.exit_code == 0 187 + self_path = journal_with_sol / "sol" / "self.md" 188 + assert self_path.read_text() == "test content\n" 189 + # No files created outside sol/ 190 + journal_files = set(f.name for f in journal_with_sol.iterdir() if f.is_file()) 191 + assert "self.md" not in journal_files 192 + 193 + def test_agency_write_stays_in_sol_dir(self, journal_with_sol): 194 + """Write to agency.md goes to sol/agency.md, not anywhere else.""" 195 + result = runner.invoke(app, ["agency", "--write"], input="test content\n") 196 + assert result.exit_code == 0 197 + agency_path = journal_with_sol / "sol" / "agency.md" 198 + assert agency_path.read_text() == "test content\n" 199 + # No files created outside sol/ 200 + journal_files = set(f.name for f in journal_with_sol.iterdir() if f.is_file()) 201 + assert "agency.md" not in journal_files 202 + 203 + 204 + class TestHeartbeatEnsureSolDirectory: 205 + """Verify the heartbeat bug fix — ensure_sol_directory() takes no args.""" 206 + 207 + def test_ensure_sol_directory_no_args(self): 208 + """ensure_sol_directory accepts no positional args (heartbeat.py:32 fix).""" 209 + import inspect 210 + 211 + from think.awareness import ensure_sol_directory 212 + 213 + sig = inspect.signature(ensure_sol_directory) 214 + params = [p for p in sig.parameters.values() if p.default is inspect.Parameter.empty] 215 + assert len(params) == 0, "ensure_sol_directory should take no required arguments" 216 + 217 + def test_heartbeat_calls_correctly(self): 218 + """heartbeat.py calls ensure_sol_directory() without arguments.""" 219 + import ast 220 + 221 + from pathlib import Path 222 + 223 + heartbeat_path = Path(__file__).parent.parent / "think" / "heartbeat.py" 224 + tree = ast.parse(heartbeat_path.read_text()) 225 + 226 + for node in ast.walk(tree): 227 + if ( 228 + isinstance(node, ast.Call) 229 + and isinstance(node.func, ast.Name) 230 + and node.func.id == "ensure_sol_directory" 231 + ): 232 + assert len(node.args) == 0, ( 233 + f"ensure_sol_directory() called with {len(node.args)} args at line {node.lineno}" 234 + )
+3 -1
think/call.py
··· 71 71 72 72 _discover_app_calls() 73 73 74 - # Mount built-in journal CLI (not auto-discovered since it lives under think/) 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.sol import app as sol_app 76 77 77 78 call_app.add_typer(journal_app, name="journal") 79 + call_app.add_typer(sol_app, name="sol") 78 80 79 81 80 82 # General-purpose navigate command (migrated from apps/chat/call.py)
+48 -3
think/heartbeat.py
··· 20 20 logger = logging.getLogger(__name__) 21 21 22 22 23 + RECENCY_WINDOW_HOURS = 12 24 + 25 + 26 + def _last_success_time(health_dir: Path) -> datetime | None: 27 + """Return the timestamp of the most recent successful heartbeat run.""" 28 + log_file = health_dir / "heartbeat.log" 29 + if not log_file.exists(): 30 + return None 31 + try: 32 + lines = log_file.read_text().strip().splitlines() 33 + except OSError: 34 + return None 35 + for line in reversed(lines): 36 + if "outcome=success" in line: 37 + ts_str = line.split()[0] 38 + try: 39 + return datetime.fromisoformat(ts_str) 40 + except ValueError: 41 + continue 42 + return None 43 + 44 + 23 45 def main() -> None: 24 46 """Entry point for ``sol heartbeat``.""" 25 47 parser = argparse.ArgumentParser( 26 48 prog="sol heartbeat", 27 49 description="Run periodic self-check agent", 28 50 ) 51 + parser.add_argument( 52 + "--force", 53 + action="store_true", 54 + help="Run full check regardless of recency", 55 + ) 29 56 args = setup_cli(parser) 30 57 31 58 journal = Path(get_journal()) 32 - ensure_sol_directory(str(journal)) 59 + ensure_sol_directory() 33 60 health_dir = journal / "health" 34 61 health_dir.mkdir(parents=True, exist_ok=True) 62 + 63 + # Recency check: skip if a recent successful run exists 64 + if not args.force: 65 + last_success = _last_success_time(health_dir) 66 + if last_success is not None: 67 + hours_since = (datetime.now() - last_success).total_seconds() / 3600 68 + if hours_since < RECENCY_WINDOW_HOURS: 69 + logger.info( 70 + "Heartbeat succeeded %.1f hours ago (within %d-hour window), skipping", 71 + hours_since, 72 + RECENCY_WINDOW_HOURS, 73 + ) 74 + sys.exit(0) 75 + 35 76 pid_file = health_dir / "heartbeat.pid" 36 77 37 78 try: ··· 62 103 pid_file.write_text(str(os.getpid())) 63 104 start_time = time.monotonic() 64 105 65 - # Spawn heartbeat agent 66 - agent_id = cortex_request(prompt="Run heartbeat check", name="heartbeat") 106 + # Spawn heartbeat agent with explicit journal path so the cogitate 107 + # agent never needs to discover it via filesystem search. 108 + agent_id = cortex_request( 109 + prompt=f"Run heartbeat check.\n\nJournal path: {journal}", 110 + name="heartbeat", 111 + ) 67 112 if agent_id is None: 68 113 logger.error("Failed to send heartbeat request to cortex") 69 114 _log_run(health_dir, start_time, "error")
+90
think/tools/sol.py
··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + """CLI commands for sol/ identity directory. 5 + 6 + Provides read and write access to ``{journal}/sol/self.md`` and 7 + ``{journal}/sol/agency.md`` — sol's identity and initiative files. 8 + 9 + Mounted by ``think.call`` as ``sol call sol ...``. 10 + """ 11 + 12 + import sys 13 + 14 + import typer 15 + 16 + from think.awareness import ensure_sol_directory, update_self_md_section 17 + 18 + app = typer.Typer(help="Sol identity directory — self.md and agency.md.") 19 + 20 + 21 + def _sol_dir(): 22 + """Return the sol/ directory path, creating it if needed.""" 23 + return ensure_sol_directory() 24 + 25 + 26 + @app.command("self") 27 + def self_cmd( 28 + write: bool = typer.Option(False, "--write", "-w", help="Write self.md from stdin."), 29 + update_section: str | None = typer.Option( 30 + None, 31 + "--update-section", 32 + help="Update a specific ## section of self.md from stdin (e.g. 'my owner').", 33 + ), 34 + ) -> None: 35 + """Read or write sol/self.md.""" 36 + sol_dir = _sol_dir() 37 + self_path = sol_dir / "self.md" 38 + 39 + if update_section: 40 + content = sys.stdin.read() 41 + if not content.strip(): 42 + typer.echo("Error: no content provided on stdin.", err=True) 43 + raise typer.Exit(1) 44 + if update_self_md_section(update_section, content.strip()): 45 + typer.echo(f"Updated ## {update_section} in self.md.") 46 + else: 47 + typer.echo(f"Error: section '## {update_section}' not found.", err=True) 48 + raise typer.Exit(1) 49 + return 50 + 51 + if write: 52 + content = sys.stdin.read() 53 + if not content.strip(): 54 + typer.echo("Error: no content provided on stdin.", err=True) 55 + raise typer.Exit(1) 56 + self_path.write_text(content, encoding="utf-8") 57 + typer.echo("self.md updated.") 58 + return 59 + 60 + # Read mode 61 + if not self_path.exists(): 62 + typer.echo("self.md not found.", err=True) 63 + raise typer.Exit(1) 64 + typer.echo(self_path.read_text(encoding="utf-8")) 65 + 66 + 67 + @app.command("agency") 68 + def agency_cmd( 69 + write: bool = typer.Option( 70 + False, "--write", "-w", help="Write agency.md from stdin." 71 + ), 72 + ) -> None: 73 + """Read or write sol/agency.md.""" 74 + sol_dir = _sol_dir() 75 + agency_path = sol_dir / "agency.md" 76 + 77 + if write: 78 + content = sys.stdin.read() 79 + if not content.strip(): 80 + typer.echo("Error: no content provided on stdin.", err=True) 81 + raise typer.Exit(1) 82 + agency_path.write_text(content, encoding="utf-8") 83 + typer.echo("agency.md updated.") 84 + return 85 + 86 + # Read mode 87 + if not agency_path.exists(): 88 + typer.echo("agency.md not found.", err=True) 89 + raise typer.Exit(1) 90 + typer.echo(agency_path.read_text(encoding="utf-8"))