personal memory agent
at main 284 lines 8.5 kB view raw
1# SPDX-License-Identifier: AGPL-3.0-only 2# Copyright (c) 2026 sol pbc 3 4"""CLI commands for sol/ identity directory. 5 6Provides read and write access to ``{journal}/sol/self.md``, 7``{journal}/sol/partner.md``, ``{journal}/sol/agency.md``, and 8``{journal}/sol/pulse.md``, and ``{journal}/sol/awareness.md`` — sol's 9identity and initiative files. Also provides read access to the morning 10briefing at 11``{journal}/YYYYMMDD/agents/morning_briefing.md``. 12 13Mounted by ``think.call`` as ``sol call identity ...``. 14""" 15 16import sys 17 18import typer 19 20from think.entities.core import atomic_write 21from think.awareness import ( 22 _log_identity_change, 23 ensure_sol_directory, 24 update_identity_section, 25 update_self_md_section, 26) 27 28app = typer.Typer( 29 help="Sol identity directory — self.md, partner.md, agency.md, pulse.md, awareness.md, and morning briefing." 30) 31 32 33def _sol_dir(): 34 """Return the sol/ directory path, creating it if needed.""" 35 return ensure_sol_directory() 36 37 38def _resolve_content(value: str | None) -> str: 39 """Return *value* if provided, else read stdin. Exit 1 if empty.""" 40 if value is not None: 41 content = value 42 else: 43 content = sys.stdin.read() 44 if not content.strip(): 45 typer.echo("Error: no content provided.", err=True) 46 raise typer.Exit(1) 47 return content 48 49 50@app.command("self") 51def self_cmd( 52 write: bool = typer.Option( 53 False, "--write", "-w", help="Overwrite self.md (content via --value or stdin)." 54 ), 55 update_section: str | None = typer.Option( 56 None, 57 "--update-section", 58 help="Update a specific ## section of self.md (content via --value or stdin).", 59 ), 60 value: str | None = typer.Option( 61 None, "--value", help="Content to write (alternative to stdin)." 62 ), 63) -> None: 64 """Read or write sol/self.md.""" 65 sol_dir = _sol_dir() 66 self_path = sol_dir / "self.md" 67 68 if update_section: 69 content = _resolve_content(value) 70 if update_self_md_section(update_section, content.strip()): 71 typer.echo(f"Updated ## {update_section} in self.md.") 72 else: 73 typer.echo(f"Error: section '## {update_section}' not found.", err=True) 74 raise typer.Exit(1) 75 return 76 77 if write: 78 content = _resolve_content(value) 79 old_content = ( 80 self_path.read_text(encoding="utf-8") if self_path.exists() else "" 81 ) 82 atomic_write(self_path, content) 83 _log_identity_change( 84 "self.md", old_content, content, section=None, source="cli" 85 ) 86 typer.echo("self.md updated.") 87 return 88 89 # Read mode 90 if not self_path.exists(): 91 typer.echo("self.md not found.", err=True) 92 raise typer.Exit(1) 93 typer.echo(self_path.read_text(encoding="utf-8")) 94 95 96@app.command("partner") 97def partner_cmd( 98 write: bool = typer.Option( 99 False, 100 "--write", 101 "-w", 102 help="Overwrite partner.md (content via --value or stdin).", 103 ), 104 update_section: str | None = typer.Option( 105 None, 106 "--update-section", 107 help="Update a specific ## section of partner.md (content via --value or stdin).", 108 ), 109 value: str | None = typer.Option( 110 None, "--value", help="Content to write (alternative to stdin)." 111 ), 112) -> None: 113 """Read or write sol/partner.md.""" 114 sol_dir = _sol_dir() 115 partner_path = sol_dir / "partner.md" 116 117 if update_section: 118 content = _resolve_content(value) 119 if update_identity_section("partner.md", update_section, content.strip()): 120 typer.echo(f"Updated ## {update_section} in partner.md.") 121 else: 122 typer.echo(f"Error: section '## {update_section}' not found.", err=True) 123 raise typer.Exit(1) 124 return 125 126 if write: 127 content = _resolve_content(value) 128 old_content = ( 129 partner_path.read_text(encoding="utf-8") if partner_path.exists() else "" 130 ) 131 atomic_write(partner_path, content) 132 _log_identity_change( 133 "partner.md", old_content, content, section=None, source="cli" 134 ) 135 typer.echo("partner.md updated.") 136 return 137 138 # Read mode 139 if not partner_path.exists(): 140 typer.echo("partner.md not found.", err=True) 141 raise typer.Exit(1) 142 typer.echo(partner_path.read_text(encoding="utf-8")) 143 144 145@app.command("agency") 146def agency_cmd( 147 write: bool = typer.Option( 148 False, 149 "--write", 150 "-w", 151 help="Overwrite agency.md (content via --value or stdin).", 152 ), 153 value: str | None = typer.Option( 154 None, "--value", help="Content to write (alternative to stdin)." 155 ), 156) -> None: 157 """Read or write sol/agency.md.""" 158 sol_dir = _sol_dir() 159 agency_path = sol_dir / "agency.md" 160 161 if write: 162 content = _resolve_content(value) 163 old_content = ( 164 agency_path.read_text(encoding="utf-8") if agency_path.exists() else "" 165 ) 166 atomic_write(agency_path, content) 167 _log_identity_change( 168 "agency.md", 169 old_content, 170 content, 171 section=None, 172 source="cli", 173 ) 174 typer.echo("agency.md updated.") 175 return 176 177 # Read mode 178 if not agency_path.exists(): 179 typer.echo("agency.md not found.", err=True) 180 raise typer.Exit(1) 181 typer.echo(agency_path.read_text(encoding="utf-8")) 182 183 184@app.command("pulse") 185def pulse_cmd( 186 write: bool = typer.Option( 187 False, 188 "--write", 189 "-w", 190 help="Overwrite pulse.md (content via --value or stdin).", 191 ), 192 value: str | None = typer.Option( 193 None, "--value", help="Content to write (alternative to stdin)." 194 ), 195) -> None: 196 """Read or write sol/pulse.md.""" 197 sol_dir = _sol_dir() 198 pulse_path = sol_dir / "pulse.md" 199 200 if write: 201 content = _resolve_content(value) 202 old_content = ( 203 pulse_path.read_text(encoding="utf-8") if pulse_path.exists() else "" 204 ) 205 atomic_write(pulse_path, content) 206 _log_identity_change( 207 "pulse.md", old_content, content, section=None, source="cli" 208 ) 209 typer.echo("pulse.md updated.") 210 return 211 212 # Read mode 213 if not pulse_path.exists(): 214 typer.echo("pulse.md not found.", err=True) 215 raise typer.Exit(1) 216 typer.echo(pulse_path.read_text(encoding="utf-8")) 217 218 219@app.command("awareness") 220def awareness_cmd( 221 write: bool = typer.Option( 222 False, 223 "--write", 224 "-w", 225 help="Overwrite awareness.md (content via --value or stdin).", 226 ), 227 value: str | None = typer.Option( 228 None, "--value", help="Content to write (alternative to stdin)." 229 ), 230) -> None: 231 """Read or write sol/awareness.md.""" 232 sol_dir = _sol_dir() 233 awareness_path = sol_dir / "awareness.md" 234 235 if write: 236 content = _resolve_content(value) 237 old_content = ( 238 awareness_path.read_text(encoding="utf-8") 239 if awareness_path.exists() 240 else "" 241 ) 242 atomic_write(awareness_path, content) 243 _log_identity_change( 244 "awareness.md", old_content, content, section=None, source="cli" 245 ) 246 typer.echo("awareness.md updated.") 247 return 248 249 # Read mode 250 if not awareness_path.exists(): 251 typer.echo("awareness.md not found.", err=True) 252 raise typer.Exit(1) 253 typer.echo(awareness_path.read_text(encoding="utf-8")) 254 255 256@app.command("briefing") 257def briefing_cmd( 258 day: str | None = typer.Option(None, "--day", "-d", help="Specific day YYYYMMDD."), 259) -> None: 260 """Read the morning briefing from YYYYMMDD/agents/morning_briefing.md.""" 261 from pathlib import Path as _Path 262 263 from think.utils import get_journal 264 265 journal = _Path(get_journal()) 266 267 if day: 268 path = journal / day / "agents" / "morning_briefing.md" 269 if not path.exists(): 270 typer.echo("No briefing found.", err=True) 271 raise typer.Exit(1) 272 typer.echo(path.read_text(encoding="utf-8")) 273 return 274 275 # No day specified — find most recent 276 agents_dirs = sorted(journal.glob("*/agents"), reverse=True) 277 for agents_dir in agents_dirs: 278 briefing = agents_dir / "morning_briefing.md" 279 if briefing.exists() and briefing.stat().st_size > 0: 280 typer.echo(briefing.read_text(encoding="utf-8")) 281 return 282 283 typer.echo("No briefing found.", err=True) 284 raise typer.Exit(1)