######## row_links
--- handlers.py
NAME = re.compile(rf"(?<![\w./-])([\w.-]+\.(?:{EXT}))(?![\w/-])")
class NameAmbiguousFiles(Handler):
    def handle(self, context: AgentContext, event: AgentMessageSent) -> None:
        told = set(context.state.get("ambiguous", []))
        project = context.record.root.parent
        for name in dict.fromkeys(m.group(1) for m in NAME.finditer(event.text)):
            found = matching(project, name) if a_file(name) and name not in told else []
            if len(found) > 1:
                told.add(name)
                context.agent.whisper(AMBIGUOUS, name=name, count=len(found), paths=", ".join(found[:5]))
        context.state.set("ambiguous", sorted(told))
--- feature.py
class RowLinks(Feature):
    details = RowLinksDetails
    def register(self, journal: Journal) -> None:
        journal.client.formatter(MarkRows())
        journal.client.formatter(MarkPaths())
        journal.client.formatter(UnwrapChips())
        journal.events.handler(NameAmbiguousFiles())
--- TEST
9:def test_a_row_named_in_text_is_a_chip_in_the_viewer_and_plain_words_everywhere_else():
39:def test_files_commits_and_links_are_marked_by_the_server_and_code_is_left_alone():
50:def test_a_file_chip_into_another_project_cannot_read_its_files():
63:def test_a_file_name_is_a_chip_when_one_project_file_has_it_and_the_agent_hears_of_a_shared_one():
######## starting_agents
--- handlers.py
LAUNCHING_FOR = 60.0
def last_conversation(record):
    rows = [row for row in Agents(record, actor=SYSTEM).rows.every()
            if row.data.get("event") and row.provider in DRIVERS and not row.subagent and not row.title.startswith(f"{row.provider}-")]
    return max(rows, key=lambda row: float(row.at), default=None)
class WakeOnMessage(Handler):
    def handle(self, context: Context, event: MessageCreated) -> None:
        if not context.settings.wake_on_message or Messages(context.record, actor=SYSTEM).load(event.n).seen[:1] != [USER]:
            return
        record = context.record
        if Sessions(record.root).holder(record.env):
            return
        environments = Environments(record, actor=SYSTEM)
        place = environments.rows.by_title(record.env)
        if place is None or time.time() - float(place.launched) < LAUNCHING_FOR:
            return
        last = last_conversation(record)
        if last is None:
            return
        launch(environments, place, last.provider, last.title)
--- feature.py
class StartingAgents(Feature):
    details = StartingAgentsDetails
    def register(self, journal: Journal) -> None:
        journal.commands.add("environment", Launch())
        journal.events.handler(WakeOnMessage())
--- launch.py
def launch(environments, place, provider: str, conversation: str = ""):
    root = environments.record.root
    detached(root, root.parent, place.title, provider, [*DRIVERS[provider].AUTO_ARGS], conversation=conversation)
    return environments.update(place.n, launched=time.time(), launched_agent=provider)
--- TEST
10:def test_a_message_wakes_the_environments_last_conversation_when_no_agent_runs(monkeypatch):
31:def test_the_agents_command_is_found_where_it_installs_itself_or_refused_in_words(tmp_path, monkeypatch):
46:def test_the_start_offers_to_carry_on_the_environments_last_session():
######## triggers
--- handlers.py
WATCHING = "watching"
CHAT_DENIED = "caught a denied word in the agent's message"
DONE = {MESSAGE: "sent a message", NUDGE: "nudged the agent", INSTRUCT: "instructed the agent", DENY: "denied the call", START: "started its sequence"}
def firing(context, text_of, from_user: bool = False) -> list:
    return [row for row in context.journal.acting(SYSTEM).get(Triggers).rows.standing()
            if (from_user or row.words_in != FROM_USER) and mentioned(row.words, text_of(str(row.words_in or "both")))]
def fire(context, agent, row, done: str = "", about: str = "") -> None:
    if row.does != START:
        context.journal.acting(SYSTEM).get(Agents).card(agent.n, label=f"Trigger {row.title} {done or DONE[row.does]}", icon=Trigger.icon,
                                                   tone="danger" if row.does == DENY else "note", title=row.wording, ref=row.ref)
    if row.does == MESSAGE:
        context.journal.acting(USER).get(Messages).create(row.title, brief=row.brief or row.text, trigger=row.n)
    elif row.does in (NUDGE, INSTRUCT):
        context.feature.journal.whisper(context.record, agent, row.does, title=row.title, text=row.wording)
    Triggers(context.record, actor=SYSTEM).fired(row.n, about)
class WatchWhatTheAgentDoes(ToolInterceptor):
    reach = Reach.MAIN
    behaviour = WATCHING
    def intercept(self, context: AgentContext, call) -> str:
        for row in firing(context, lambda scope: searched(call, scope)):
            fire(context, context.agent.row, row)
            if row.does == DENY:
                return f"{row.title} - {row.wording or 'this call is denied by a trigger'}"
        return ""
class WatchWhatTheAgentWrites(Handler):
    behaviour = WATCHING
    def handle(self, context: AgentContext, event: AgentMessageSent) -> None:
        for row in firing(context, lambda scope: "" if scope == COMMANDS else event.text):
            fire(context, context.agent.row, row, CHAT_DENIED if row.does == DENY else "")
            if row.does == DENY:
                context.agent.whisper("denied", title=row.title, text=row.wording)
class WatchWhatTheUserWrites(Handler):
    behaviour = WATCHING
    def handle(self, context: Context, event: MessageCreated) -> None:
        if event.actor != USER:
            return
        agent = context.journal.acting(SYSTEM).get(Agents).primary()
        message = context.journal.get(Messages).load(event.n)
        if message.data.get("trigger"):
            return
        text = f"{message.title} {message.brief}"
        for row in firing(context, lambda scope: "" if scope == COMMANDS else text, from_user=True):
            if agent and row.does != DENY:
                fire(context, agent, row, about=f"message:{message.n}")
--- feature.py
__all__ = ["Triggers"]
class TriggersFeature(Feature):
    details = TriggersDetails
    def register(self, journal: Journal) -> None:
        journal.events.handler(WatchWhatTheUserWrites())
        journal.events.handler(WatchWhatTheAgentWrites())
        journal.agent.interceptor(WatchWhatTheAgentDoes())
--- TEST
22:def test_a_trigger_denies_a_command_and_nudges_on_a_word():
38:def test_a_trigger_fires_on_what_the_user_writes():
47:def test_a_trigger_message_does_not_fire_the_trigger_again():
58:def test_a_trigger_fires_on_what_the_agent_says_in_the_chat():
######## terminal
--- routes.py
@dataclass(frozen=True)
class TerminalQuery(Loaded):
    level: str = EVERYTHING
    after: float = 0.0
@handles("GET", "/api/{env}/agent/{n}/terminal")
def get_terminal(req: Request) -> Reply:
    asked = req.query_as(TerminalQuery)
    level = asked.level
    if level not in TERMINAL_LEVELS:
        raise Refused(f"level is one of {', '.join(TERMINAL_LEVELS)}")
    record = req.record()
    return Reply(200, {"lines": terminal_lines(record, Agents(record, actor=USER).load(req.params["n"]).title, level, asked.after)})
--- feature.py
class KeepCommands(Handler):
    def handle(self, context: AgentContext, event: CommandRan) -> None:
        kept(context.record, context.agent.row.title, event)
class Terminal(Feature):
    details = TerminalDetails
    def register(self, journal: Journal) -> None:
        journal.routes.add(get_terminal)
        journal.events.handler(KeepCommands())
--- log.py
LOG = "terminal"
COMMANDS, JOURNAL, EVERYTHING = "commands", "journal", "everything"
LEVELS = (COMMANDS, JOURNAL, EVERYTHING)
KEPT = 200
MOST_LINES = 60
MOST_CHARS = 4000
SLACK = 20
CHANGED_FOLDER = re.compile(r"^cd\s+\S+\s*&&\s*")
JOURNAL_FOLDER = re.compile(r"(^|[\s'\"=/])\.journal(/|\s|$)")
def capped(text: str) -> str:
    lines = text.splitlines()
    if len(lines) <= MOST_LINES + SLACK and len(text) <= MOST_CHARS:
        return text
    kept = text[:MOST_CHARS].splitlines()[:MOST_LINES]
    return "\n".join([*kept, f"… {len(lines) - len(kept)} more lines were not kept"])
def journals_own(line: CommandRan) -> bool:
    command = line.command.strip()
    return line.tool == SHELL and bool(JOURNAL_CALL.match(CHANGED_FOLDER.sub("", command)) or JOURNAL_FOLDER.search(command))
def level_of(line: CommandRan) -> str:
    if line.tool in (TYPED, NOTED):
        return COMMANDS
    if line.tool == DELIVERED:
        return JOURNAL
    if line.tool == SHELL:
        return JOURNAL if journals_own(line) else COMMANDS
    return EVERYTHING
def kept(record, session: str, line: CommandRan) -> None:
    level = level_of(line)
    with record.state(f"{LOG}-{level}", session).changing() as held:
        held["lines"] = [*held.get("lines", []), {**asdict(line), "output": capped(line.output.strip()), "level": level}][-KEPT:]
def lines(record, session: str, level: str, after: float = 0.0) -> list[dict]:
    included = LEVELS[:LEVELS.index(level) + 1]
    found = (line for name in included for line in record.state(f"{LOG}-{name}", session).get("lines", []) if line["at"] > after)
    return sorted(found, key=lambda line: line["at"])[-KEPT:]
--- TEST
14:def test_shell_commands_are_kept_with_their_output_and_journal_calls_are_left_out():
28:def test_a_poll_after_the_last_line_gets_only_the_newer_ones():
36:def test_a_command_from_the_terminal_view_is_typed_into_the_agents_terminal_as_a_shell_command(monkeypatch):
######## work_modes
--- routes.py
@handles("POST", "/api/{env}/mode")
def post_mode(req: Request) -> Reply:
    from features.work_modes.modes import pick
    return Reply(200, {"mode": pick(Record(req.root, req.params["env"]), str(req.body.get("mode", "")), USER)})
--- feature.py
class WorkModes(Feature):
    details = WorkModesDetails
    def register(self, journal: Journal) -> None:
        journal.routes.add(post_mode)
        START_PARTS.add(self, carried, key=MODE)
        journal.agent.canceler(RefuseDispatchInSolo())
        journal.agent.interceptor(RefuseHelperInSolo())
        journal.agent.interceptor(RemindOrchestrator())
--- interceptors.py
REFUSED = "the user set this environment to solo: do the work yourself, with no subagents and no helpers"
class RefuseDispatchInSolo(Canceler):
    reach = Reach.MAIN
    event = DISPATCHING
    def cancel(self, context: AgentContext, data) -> str:
        return REFUSED if mode_of(context.record) == SOLO else ""
class RefuseHelperInSolo(ToolInterceptor):
    reach = Reach.MAIN
    def intercept(self, context: AgentContext, call) -> str:
        shell = call.shell_command
        if not shell or mode_of(context.record) != SOLO:
            return ""
        return REFUSED if any(made.names("helper", "dispatch") for made in calls(shell)) else ""
class RemindOrchestrator(ToolInterceptor):
    reach = Reach.MAIN
    refuses = False
    def intercept(self, context: AgentContext, call) -> str:
        if not isinstance(call, WriteCall) or mode_of(context.record) != ORCHESTRATOR:
            return ""
        if Path(call.file_path).resolve().is_relative_to(context.record.root.resolve()):
            return ""
        edits = int(context.state.get("edits", 0)) + 1
        if edits < int(context.settings.drift_after):
            context.state.set("edits", edits)
            return ""
        context.state.set("edits", 0)
        context.agent.whisper("drifted", edits=edits)
        return ""
--- TEST
23:def test_a_picked_mode_is_kept_told_to_the_agent_and_carried_into_every_start():
37:def test_solo_refuses_a_subagent_and_a_helper_and_hands_on_lets_both_through():
48:def test_orchestrator_reminds_the_agent_after_its_own_edits_and_hands_on_never(monkeypatch):
######## update_reports
--- handlers.py
COMMITS = "close_from_commits"
OFFERED = "update-offered"
OFFER_EVERY = 3600
OFFER = "offer"
class OfferAnUpdate(Handler):
    def handle(self, context: AgentContext, event: TurnStopped) -> None:
        commit = context.record.event_log.cursor_text(COMMITS)
        seen, _, at = context.record.event_log.cursor_text(OFFERED).partition(" ")
        if not commit or commit == seen:
            return
        if not seen:
            context.record.event_log.set_cursor_text(OFFERED, f"{commit} {time.time()}")
            return
        if time.time() - float(at) < OFFER_EVERY or self._unread(context):
            context.record.event_log.set_cursor_text(OFFERED, f"{commit} {at}")
            return
        context.record.event_log.set_cursor_text(OFFERED, f"{commit} {time.time()}")
        context.agent.say(OFFER)
    @staticmethod
    def _unread(context: AgentContext) -> bool:
        return any(r.data.get("kind") == UPDATE and not r.data.get("dismissed") and USER not in r.seen
                   for r in context.journal.get(Reports).rows.standing())
--- feature.py
class UpdateReports(Feature):
    details = UpdateReportsDetails
    def register(self, journal: Journal) -> None:
        for command in (Changes(), Recap(), Note(), AddItem(), Drop(), Dismiss()):
            journal.commands.add("report", command)
        journal.events.handler(OfferAnUpdate())
--- TEST
12:def test_an_update_covers_what_happened_since_the_user_opened_the_last_one():
32:def test_notes_items_and_drops_change_only_an_update():
58:def test_asking_for_an_update_starts_writing_one_and_the_agent_saying_it_does_not():
83:def test_a_commit_reminds_the_agent_it_may_write_an_update_at_most_hourly():
99:def test_an_update_report_never_starts_the_sequence_for_research_reports():
######## organization
--- handlers.py
class StartNextForGlobalRole(Handler):
    def handle(self, context: Context, event: TodoCompleted) -> None:
        todo = context.journal.get(Todos).load(event.n)
        role = role_of(context.record, todo)
        if role:
            for env, waiting in next_in_line(context.record, todo.data["domain"], todo.data["role"], context.record.env, todo.n):
                place = Record(context.record.root, env)
                Todos(place, actor=SYSTEM).unblock(waiting.n)
                started = start_role_agent(place, role, waiting.n, f"{waiting.title}\n{waiting.brief}")
                if started:
                    Todos(place, actor=SYSTEM).update(waiting.n, role_environment=started)
        if todo.data.get("role_environment") and not (role and role.cardinality == PLAN):
            stop_in(context.record, todo.data["role_environment"])
--- feature.py
class OrganizationFeature(Feature):
    details = OrganizationDetails
    def register(self, journal: Journal) -> None:
        journal.commands.add("ticket", ShowOrganization())
        journal.commands.add("todo", Delegate())
        journal.commands.intercept("todo.update", ReportCoversOutputs())
        journal.events.handler(StartNextForGlobalRole())
--- TEST
14:def test_the_organization_is_read_from_domain_and_role_files_and_refuses_what_does_not_fit(tmp_path):
37:def test_a_task_is_delegated_to_a_role_queues_behind_its_own_and_is_reported_against_its_outputs(tmp_path):
64:def test_a_role_with_a_browser_is_pointed_at_its_tickets_app():
73:def test_the_board_counts_each_role_by_the_tickets_where_its_work_is_in_hand():
101:def test_a_global_role_takes_one_task_at_a_time_across_every_environment():
123:def test_a_role_that_runs_as_an_agent_is_started_in_the_tickets_worktree(monkeypatch):
######## runtime_cleanup
--- handlers.py
class TidyRuntime(Handler):
    behaviour = WHOLE_FEATURE
    def handle(self, context: AgentContext, event: ClockTicked) -> None:
        tidy(context.record.root, context.settings.days)
@dataclass(frozen=True)
class NotificationCreated(ResourceEvent):
    on: ClassVar[str] = "notification.created"
class TidyAfterUpdate(Handler):
    def handle(self, context: Context, event: NotificationCreated) -> None:
        if context.journal.get(Notifications).load(event.n).data.get("kind") == KIND:
            tidy_files(context.record.root, context.settings.days)
--- feature.py
class RuntimeCleanup(Feature):
    details = RuntimeCleanupDetails
    def register(self, journal: Journal) -> None:
        journal.events.handler(TidyRuntime())
        journal.events.handler(TidyAfterUpdate())
--- tidy.py
TAILS = {"sessions/*/printed": 64 * 1024, "sessions/*/screen": 1024 * 1024, "*.log": 1024 * 1024, "channels/*.jsonl": 1024 * 1024}
EVENTS_KEPT = 100
READERS_WITHIN = DAY
STAGING_FOR = 3600
OUTPUTS_FOR = DAY
ARCHIVES_FOR = 90 * DAY
PROFILES_KEPT = 50
FOLDS_FOR = 7 * DAY
@dataclass(frozen=True)
class Tidied:
    removed: int = 0
    trimmed: int = 0
    events: int = 0
    leftovers: int = 0
    @property
    def summary(self) -> str:
        parts = [f"{plural(self.trimmed, 'log')} cut to their tail" if self.trimmed else "",
                 f"{plural(self.removed, 'quiet session folder')} removed" if self.removed else "",
                 f"{plural(self.events, 'old event')} dropped" if self.events else "",
                 f"{plural(self.leftovers, 'leftover')} removed" if self.leftovers else ""]
        found = "; ".join(part for part in parts if part)
        return found if found else "nothing to tidy"
def tidy(root: Path, days: float) -> Tidied:
    events = sum(Record(Path(root), env.name).event_log.trim(EVENTS_KEPT, time.time() - READERS_WITHIN)
                 for env in (Path(root) / "environments").glob("*") if (env / "events.jsonl").is_file())
    return replace(tidy_files(Path(root), days), events=events)
def tidy_files(root: Path, days: float) -> Tidied:
    left = leftovers(root)
    kept = folder(root)
    if not kept.is_dir():
        return Tidied(leftovers=left)
    quiet = time.time() - days * DAY
    removed = [d for d in sessions(root).glob("*") if d.is_dir() and max((mtime(f) / 1e9 for f in d.iterdir()), default=0) < quiet]
    for d in removed:
        shutil.rmtree(d, ignore_errors=True)
    trimmed = [f for pattern, keep in TAILS.items() for f in kept.glob(pattern) if f.is_file() and trim(f, keep)]
    return Tidied(removed=len(removed), trimmed=len(trimmed), leftovers=left)
def older(paths, age: float) -> list[Path]:
    now = time.time()
    return [path for path in paths if mtime(path) and now - mtime(path) / 1e9 > age]
def leftovers(root: Path) -> int:
    staged = [d for d in older((root / "plugins").glob(".staging-*"), STAGING_FOR) if d.is_dir()]
    archived = [f for f in older((root / "attic").glob("*.tar.gz"), ARCHIVES_FOR) if not f.name.startswith("before-")]
    unkept = older((folder(root) / "outputs").glob("output-*"), OUTPUTS_FOR)
    profiled = sorted(profiles(root).glob("*.txt"), key=mtime, reverse=True)[PROFILES_KEPT:]
    folds = older(FOLD_CACHE.glob("*.pickle"), FOLDS_FOR)
    for d in staged:
        shutil.rmtree(d, ignore_errors=True)
    for f in archived + unkept + profiled + folds:
        f.unlink(missing_ok=True)
    return len(staged) + len(archived) + len(unkept) + len(profiled) + len(folds)
def trim(f: Path, keep: int) -> bool:
    if f.stat().st_size <= keep:
        return False
    with f.open("r+b") as held:
        held.seek(-keep, 2)
        tail = held.read()
        held.seek(0)
        held.write(tail)
        held.truncate()
    return True
--- TEST
18:def test_captures_are_cut_to_their_tail_and_quiet_sessions_are_removed_whole(monkeypatch):
54:def test_the_days_to_keep_is_a_setting():
66:def test_it_runs_on_its_own_on_the_engines_clock():
75:def test_the_event_log_keeps_the_last_hundred_and_whatever_a_live_reader_has_not_reached():
90:def test_leftover_plugin_checkouts_and_old_environment_archives_are_removed_and_snapshots_kept():
141:def test_an_installed_update_tidies_at_once():
######## rules
--- handlers.py
@dataclass(frozen=True)
class RuleChanged(ResourceEvent):
    on: ClassVar[str] = "rule"
class InjectRules(Handler):
    def handle(self, context: Context, event: RuleChanged) -> None:
        brief(context.record.root.parent, context.record)
class ReviewNewRule(Handler):
    def handle(self, context: Context, event: RuleCreated) -> None:
        if event.actor != AGENT:
            return
        agent = context.journal.get(Agents).primary()
        if agent:
            context.speaking_to(agent).agent.whisper("review", n=event.n, title=context.journal.get(Rules).load(event.n).title)
--- feature.py
class RulesFeature(Feature):
    details = RulesDetails
    def register(self, journal: Journal) -> None:
        register_recital(journal, Rules)
        journal.events.handler(InjectRules())
        journal.events.handler(ReviewNewRule())
--- TEST
10:def test_a_command_that_touches_a_rules_keyword_is_whispered_the_rule_once_per_session():
56:def test_standing_rules_are_repeated_at_every_quarter_and_a_struck_rule_drops_out():
71:def test_inject_and_uninject_change_the_rules_in_the_one_block_and_pin_notices_the_rule():
98:def test_the_briefing_retires_the_old_blocks_and_leaves_a_file_it_cannot_read_safely():
136:def test_a_rule_the_agent_makes_is_sent_back_to_be_read_as_a_ruling_for_the_whole_project():
