tenseleyflow/loader / c6feb00

+import sys

+from .options import inject_resume_target

+@click.option("--resume-target", hidden=True, default=None)

+def cli(

+    resume_target: str | None,

+        model, select_model, backend, yes, permission_mode, react, no_context, plan, clarify, resume_target, no_recover,

+def main() -> None:

+    """Entry-point wrapper that supports `--resume [session-id]` syntax."""

+

+    cli.main(args=inject_resume_target(sys.argv[1:]), prog_name="loader")

+

+

+    resume_target: str | None,

+    resumed = False

+    if resume_target is not None:

+        session_id = None if resume_target == "__latest__" else resume_target

+        resumed = agent.resume_session(session_id)

+        if not resumed and session_id is None:

+            console.print("[yellow]No previous session found; starting a new session.[/yellow]")

+        elif not resumed:

+            console.print(f"[red]Session not found:[/red] {session_id}")

+            return

+        else:

+            console.print(f"[dim]Resumed session: {agent.session.session_id}[/dim]")

+            f"Workflow: {format_workflow_mode(agent.workflow_mode)}",

+            f"Permissions: {format_permission_mode(agent.active_permission_mode)}",

+            f"Workflow: {format_workflow_mode(agent.workflow_mode)}",

+            f"Permissions: {format_permission_mode(agent.active_permission_mode)}",

+            workflow_mode=agent.workflow_mode,

+            permission_mode=agent.active_permission_mode,

+"""Reusable CLI argument helpers for Loader."""

+

+from __future__ import annotations

+

+

+def inject_resume_target(argv: list[str]) -> list[str]:

+    """Rewrite `--resume [session-id]` into a hidden click option."""

+

+    rewritten: list[str] = []

+    index = 0

+    while index < len(argv):

+        arg = argv[index]

+        if arg != "--resume":

+            rewritten.append(arg)

+            index += 1

+            continue

+

+        next_arg = argv[index + 1] if index + 1 < len(argv) else None

+        if next_arg and not next_arg.startswith("-"):

+            rewritten.extend(["--resume-target", next_arg])

+            index += 2

+        else:

+            rewritten.extend(["--resume-target", "__latest__"])

+            index += 1

+    return rewritten

+from .memory import MemoryStore

+        if summary.definition_of_done and summary.definition_of_done.status == "done":

+            MemoryStore(self.agent.project_root).capture_definition_of_done(

+                build_verification_summary(summary.definition_of_done.evidence)

+            )

+from .memory import MemoryStore

+class MemoryLifecycleHook(BaseToolHook):

+    """Mirror durable memory updates into the session notepad."""

+

+    async def post_tool_use(self, context: HookContext) -> HookResult:

+        if context.result is None or context.result.is_error:

+            return HookResult()

+

+        store = MemoryStore(context.permission_policy.workspace_root)

+        if context.tool_call.name == "project_memory_add_note":

+            category = str(context.tool_call.arguments.get("category", "")).strip()

+            content = str(context.tool_call.arguments.get("content", "")).strip()

+            if category and content:

+                store.append_notepad_working(

+                    f"Remembered note [{category}]: {content}"

+                )

+        elif context.tool_call.name == "project_memory_add_directive":

+            directive = str(context.tool_call.arguments.get("directive", "")).strip()

+            priority = str(

+                context.tool_call.arguments.get("priority", "normal")

+            ).strip()

+            if directive:

+                store.append_notepad_working(

+                    f"Remembered directive [{priority}]: {directive}"

+                )

+        return HookResult()

+

+

+            MemoryLifecycleHook(),

+"""Durable project memory and working-notepad storage under `.loader/`."""

+

+from __future__ import annotations

+

+import json

+from dataclasses import dataclass

+from datetime import UTC, datetime

+from pathlib import Path

+from typing import Any

+

+

+def _utc_now() -> str:

+    return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")

+

+

+@dataclass(slots=True)

+class ProjectMemoryNote:

+    """Categorized project-memory note."""

+

+    category: str

+    content: str

+    timestamp: str

+

+    def to_dict(self) -> dict[str, str]:

+        return {

+            "category": self.category,

+            "content": self.content,

+            "timestamp": self.timestamp,

+        }

+

+

+@dataclass(slots=True)

+class ProjectMemoryDirective:

+    """Persistent user or project directive."""

+

+    directive: str

+    priority: str

+    context: str | None

+    timestamp: str

+

+    def to_dict(self) -> dict[str, str | None]:

+        return {

+            "directive": self.directive,

+            "priority": self.priority,

+            "context": self.context,

+            "timestamp": self.timestamp,

+        }

+

+

+class MemoryStore:

+    """Manage project memory and the working notepad."""

+

+    def __init__(self, project_root: Path) -> None:

+        self.project_root = project_root

+        self.loader_root = project_root / ".loader"

+        self.project_memory_path = self.loader_root / "project-memory.json"

+        self.notepad_path = self.loader_root / "notepad.md"

+

+    def ensure_layout(self) -> None:

+        """Ensure the `.loader/` directory exists."""

+

+        self.loader_root.mkdir(parents=True, exist_ok=True)

+

+    def read_project_memory(self, section: str = "all") -> Any:

+        """Read the full project memory or one named section."""

+

+        memory = self._load_project_memory()

+        if section == "all":

+            return memory

+        return memory.get(section)

+

+    def write_project_memory(

+        self,

+        memory: dict[str, Any],

+        *,

+        merge: bool = True,

+    ) -> dict[str, Any]:

+        """Write or merge project memory state."""

+

+        existing = self._load_project_memory() if merge else {}

+        merged = {**existing, **memory}

+        self._save_project_memory(merged)

+        return merged

+

+    def add_project_note(self, category: str, content: str) -> dict[str, Any]:

+        """Append one categorized note to project memory."""

+

+        memory = self._load_project_memory()

+        notes = list(memory.get("notes", []))

+        note = ProjectMemoryNote(

+            category=category.strip(),

+            content=content.strip(),

+            timestamp=_utc_now(),

+        )

+        notes.append(note.to_dict())

+        memory["notes"] = notes

+        self._save_project_memory(memory)

+        return note.to_dict()

+

+    def add_project_directive(

+        self,

+        directive: str,

+        *,

+        priority: str = "normal",

+        context: str | None = None,

+    ) -> dict[str, Any]:

+        """Append one durable directive to project memory."""

+

+        memory = self._load_project_memory()

+        directives = list(memory.get("directives", []))

+        directive_entry = ProjectMemoryDirective(

+            directive=directive.strip(),

+            priority=(priority or "normal").strip(),

+            context=context.strip() if context else None,

+            timestamp=_utc_now(),

+        )

+        directives.append(directive_entry.to_dict())

+        memory["directives"] = directives

+        self._save_project_memory(memory)

+        return directive_entry.to_dict()

+

+    def read_notepad(self, section: str = "all") -> str:

+        """Read the full notepad or one named section."""

+

+        sections = self._load_notepad_sections()

+        if section == "all":

+            return self._render_notepad(sections)

+        return sections[section]

+

+    def write_notepad_priority(self, content: str) -> str:

+        """Replace the priority context section."""

+

+        sections = self._load_notepad_sections()

+        sections["priority"] = content.strip()

+        self._save_notepad_sections(sections)

+        return sections["priority"]

+

+    def append_notepad_working(self, content: str) -> str:

+        """Append one timestamped working-memory entry."""

+

+        sections = self._load_notepad_sections()

+        entry = f"- [{_utc_now()}] {content.strip()}"

+        sections["working"] = self._append_markdown_entry(sections["working"], entry)

+        self._save_notepad_sections(sections)

+        return entry

+

+    def append_notepad_manual(self, content: str) -> str:

+        """Append one manual note that should not be auto-pruned."""

+

+        sections = self._load_notepad_sections()

+        entry = f"- {content.strip()}"

+        sections["manual"] = self._append_markdown_entry(sections["manual"], entry)

+        self._save_notepad_sections(sections)

+        return entry

+

+    def capture_definition_of_done(self, evidence_summary: str) -> dict[str, Any] | None:

+        """Persist a useful evidence summary into project memory."""

+

+        normalized = evidence_summary.strip()

+        if not normalized or normalized == "Verification: skipped (no evidence required).":

+            return None

+        return self.add_project_note("definition_of_done", normalized)

+

+    def _load_project_memory(self) -> dict[str, Any]:

+        self.ensure_layout()

+        if not self.project_memory_path.exists():

+            return {}

+        try:

+            raw = json.loads(self.project_memory_path.read_text())

+        except json.JSONDecodeError:

+            return {}

+        return raw if isinstance(raw, dict) else {}

+

+    def _save_project_memory(self, memory: dict[str, Any]) -> None:

+        self.ensure_layout()

+        self.project_memory_path.write_text(json.dumps(memory, indent=2, sort_keys=True))

+

+    def _load_notepad_sections(self) -> dict[str, str]:

+        self.ensure_layout()

+        sections = {

+            "priority": "",

+            "working": "",

+            "manual": "",

+        }

+        if not self.notepad_path.exists():

+            return sections

+

+        current: str | None = None

+        for line in self.notepad_path.read_text().splitlines():

+            if line == "## Priority Context":

+                current = "priority"

+                continue

+            if line == "## Working Memory":

+                current = "working"

+                continue

+            if line == "## Manual Notes":

+                current = "manual"

+                continue

+            if current is None or line == "# Loader Notepad":

+                continue

+            sections[current] = (

+                f"{sections[current]}\n{line}".strip()

+                if sections[current]

+                else line

+            )

+        return sections

+

+    def _save_notepad_sections(self, sections: dict[str, str]) -> None:

+        self.ensure_layout()

+        self.notepad_path.write_text(self._render_notepad(sections))

+

+    @staticmethod

+    def _append_markdown_entry(existing: str, entry: str) -> str:

+        return f"{existing}\n{entry}".strip() if existing else entry

+

+    @staticmethod

+    def _render_notepad(sections: dict[str, str]) -> str:

+        return "\n".join(

+            [

+                "# Loader Notepad",

+                "",

+                "## Priority Context",

+                sections.get("priority", "").strip() or "_Empty_",

+                "",

+                "## Working Memory",

+                sections.get("working", "").strip() or "_Empty_",

+                "",

+                "## Manual Notes",

+                sections.get("manual", "").strip() or "_Empty_",

+                "",

+            ]

+        )

+    from .memory_tools import (

+        NotepadReadTool,

+        NotepadWriteManualTool,

+        NotepadWritePriorityTool,

+        NotepadWriteWorkingTool,

+        ProjectMemoryAddDirectiveTool,

+        ProjectMemoryAddNoteTool,

+        ProjectMemoryReadTool,

+        ProjectMemoryWriteTool,

+    )

+    registry.register(ProjectMemoryReadTool())

+    registry.register(ProjectMemoryWriteTool())

+    registry.register(ProjectMemoryAddNoteTool())

+    registry.register(ProjectMemoryAddDirectiveTool())

+    registry.register(NotepadReadTool())

+    registry.register(NotepadWritePriorityTool())

+    registry.register(NotepadWriteWorkingTool())

+    registry.register(NotepadWriteManualTool())

+"""Native Loader tools for project memory and working notes."""

+

+from __future__ import annotations

+

+import json

+from pathlib import Path

+from typing import Any

+

+from ..runtime.memory import MemoryStore

+from ..runtime.permissions import PermissionMode

+from .base import Tool, ToolResult

+

+

+class MemoryTool(Tool):

+    """Shared base class for `.loader/` memory tools."""

+

+    def __init__(self, workspace_root: Path | str | None = None) -> None:

+        self.workspace_root = (

+            Path(workspace_root).expanduser().resolve() if workspace_root else None

+        )

+

+    def set_workspace_root(self, workspace_root: Path | None) -> None:

+        self.workspace_root = workspace_root

+

+    def store(self) -> MemoryStore:

+        return MemoryStore(self.workspace_root or Path.cwd())

+

+

+class ProjectMemoryReadTool(MemoryTool):

+    """Read project memory."""

+

+    required_permission = PermissionMode.READ_ONLY

+

+    @property

+    def name(self) -> str:

+        return "project_memory_read"

+

+    @property

+    def description(self) -> str:

+        return "Read project memory from .loader/project-memory.json."

+

+    @property

+    def parameters(self) -> dict[str, Any]:

+        return {

+            "type": "object",

+            "properties": {

+                "section": {

+                    "type": "string",

+                    "enum": [

+                        "all",

+                        "techStack",

+                        "build",

+                        "conventions",

+                        "structure",

+                        "notes",

+                        "directives",

+                    ],

+                    "description": "Optional project-memory section to read.",

+                }

+            },

+        }

+

+    async def execute(self, section: str = "all", **kwargs: Any) -> ToolResult:

+        payload = self.store().read_project_memory(section=section or "all")

+        return ToolResult(

+            output=json.dumps(payload, indent=2, sort_keys=True),

+            metadata={"section": section or "all", "payload": payload},

+        )

+

+

+class ProjectMemoryWriteTool(MemoryTool):

+    """Write project memory."""

+

+    required_permission = PermissionMode.WORKSPACE_WRITE

+

+    @property

+    def name(self) -> str:

+        return "project_memory_write"

+

+    @property

+    def description(self) -> str:

+        return "Write or merge project memory in .loader/project-memory.json."

+

+    @property

+    def parameters(self) -> dict[str, Any]:

+        return {

+            "type": "object",

+            "properties": {

+                "memory": {

+                    "type": "object",

+                    "description": "Memory object to write.",

+                },

+                "merge": {

+                    "type": "boolean",

+                    "description": "Merge with existing memory when true.",

+                },

+            },

+            "required": ["memory"],

+        }

+

+    async def execute(

+        self,

+        memory: dict[str, Any],

+        merge: bool = True,

+        **kwargs: Any,

+    ) -> ToolResult:

+        payload = self.store().write_project_memory(memory, merge=merge)

+        return ToolResult(

+            output=json.dumps(payload, indent=2, sort_keys=True),

+            metadata={"memory": payload, "merge": merge},

+        )

+

+

+class ProjectMemoryAddNoteTool(MemoryTool):

+    """Append a categorized note to project memory."""

+

+    required_permission = PermissionMode.WORKSPACE_WRITE

+

+    @property

+    def name(self) -> str:

+        return "project_memory_add_note"

+

+    @property

+    def description(self) -> str:

+        return "Add a categorized note to project memory."

+

+    @property

+    def parameters(self) -> dict[str, Any]:

+        return {

+            "type": "object",

+            "properties": {

+                "category": {"type": "string"},

+                "content": {"type": "string"},

+            },

+            "required": ["category", "content"],

+        }

+

+    async def execute(self, category: str, content: str, **kwargs: Any) -> ToolResult:

+        payload = self.store().add_project_note(category, content)

+        return ToolResult(

+            output=json.dumps(payload, indent=2, sort_keys=True),

+            metadata=payload,

+        )

+

+

+class ProjectMemoryAddDirectiveTool(MemoryTool):

+    """Append a persistent directive to project memory."""

+

+    required_permission = PermissionMode.WORKSPACE_WRITE

+

+    @property

+    def name(self) -> str:

+        return "project_memory_add_directive"

+

+    @property

+    def description(self) -> str:

+        return "Add a durable directive to project memory."

+

+    @property

+    def parameters(self) -> dict[str, Any]:

+        return {

+            "type": "object",

+            "properties": {

+                "directive": {"type": "string"},

+                "priority": {

+                    "type": "string",

+                    "enum": ["high", "normal"],

+                },

+                "context": {"type": "string"},

+            },

+            "required": ["directive"],

+        }

+

+    async def execute(

+        self,

+        directive: str,

+        priority: str = "normal",

+        context: str | None = None,

+        **kwargs: Any,

+    ) -> ToolResult:

+        payload = self.store().add_project_directive(

+            directive,

+            priority=priority,

+            context=context,

+        )

+        return ToolResult(

+            output=json.dumps(payload, indent=2, sort_keys=True),

+            metadata=payload,

+        )

+

+

+class NotepadReadTool(MemoryTool):

+    """Read Loader's durable working notepad."""

+

+    required_permission = PermissionMode.READ_ONLY

+

+    @property

+    def name(self) -> str:

+        return "notepad_read"

+

+    @property

+    def description(self) -> str:

+        return "Read the Loader notepad from .loader/notepad.md."

+

+    @property

+    def parameters(self) -> dict[str, Any]:

+        return {

+            "type": "object",

+            "properties": {

+                "section": {

+                    "type": "string",

+                    "enum": ["all", "priority", "working", "manual"],

+                }

+            },

+        }

+

+    async def execute(self, section: str = "all", **kwargs: Any) -> ToolResult:

+        payload = self.store().read_notepad(section=section or "all")

+        return ToolResult(

+            output=payload,

+            metadata={"section": section or "all", "payload": payload},

+        )

+

+

+class NotepadWritePriorityTool(MemoryTool):

+    """Replace the priority context section in the notepad."""

+

+    required_permission = PermissionMode.WORKSPACE_WRITE

+

+    @property

+    def name(self) -> str:

+        return "notepad_write_priority"

+

+    @property

+    def description(self) -> str:

+        return "Replace the priority-context section in .loader/notepad.md."

+

+    @property

+    def parameters(self) -> dict[str, Any]:

+        return {

+            "type": "object",

+            "properties": {"content": {"type": "string"}},

+            "required": ["content"],

+        }

+

+    async def execute(self, content: str, **kwargs: Any) -> ToolResult:

+        payload = self.store().write_notepad_priority(content)

+        return ToolResult(output=payload, metadata={"content": payload})

+

+

+class NotepadWriteWorkingTool(MemoryTool):

+    """Append one timestamped working-memory note."""

+

+    required_permission = PermissionMode.WORKSPACE_WRITE

+

+    @property

+    def name(self) -> str:

+        return "notepad_write_working"

+

+    @property

+    def description(self) -> str:

+        return "Append a timestamped entry to the working-memory section."

+

+    @property

+    def parameters(self) -> dict[str, Any]:

+        return {

+            "type": "object",

+            "properties": {"content": {"type": "string"}},

+            "required": ["content"],

+        }

+

+    async def execute(self, content: str, **kwargs: Any) -> ToolResult:

+        payload = self.store().append_notepad_working(content)

+        return ToolResult(output=payload, metadata={"content": payload})

+

+

+class NotepadWriteManualTool(MemoryTool):

+    """Append one manual note."""

+

+    required_permission = PermissionMode.WORKSPACE_WRITE

+

+    @property

+    def name(self) -> str:

+        return "notepad_write_manual"

+

+    @property

+    def description(self) -> str:

+        return "Append a manual note to .loader/notepad.md."

+

+    @property

+    def parameters(self) -> dict[str, Any]:

+        return {

+            "type": "object",

+            "properties": {"content": {"type": "string"}},

+            "required": ["content"],

+        }

+

+    async def execute(self, content: str, **kwargs: Any) -> ToolResult:

+        payload = self.store().append_notepad_manual(content)

+        return ToolResult(output=payload, metadata={"content": payload})

+"""Tests for CLI resume argument rewriting."""

+

+from __future__ import annotations

+

+from loader.cli.options import inject_resume_target

+

+

+def test_inject_resume_target_supports_flag_and_named_session() -> None:

+    assert inject_resume_target([]) == []

+    assert inject_resume_target(["--resume"]) == ["--resume-target", "__latest__"]

+    assert inject_resume_target(["--resume", "session-123"]) == [

+        "--resume-target",

+        "session-123",

+    ]

+    assert inject_resume_target(["--resume", "session-123", "fix runtime"]) == [

+        "--resume-target",

+        "session-123",

+        "fix runtime",

+    ]

+"""Tests for project memory and notepad tools."""

+

+from __future__ import annotations

+

+import json

+from pathlib import Path

+

+import pytest

+

+from loader.agent.loop import AgentConfig

+from loader.llm.base import CompletionResponse, ToolCall

+from loader.runtime.executor import ToolExecutionState, ToolExecutor

+from loader.runtime.hooks import HookManager, MemoryLifecycleHook

+from loader.runtime.permissions import PermissionMode, build_permission_policy

+from loader.runtime.tracing import RuntimeTracer

+from loader.tools.base import create_default_registry

+from tests.helpers.runtime_harness import ScriptedBackend, run_scenario

+

+

+@pytest.mark.asyncio

+async def test_project_memory_tools_round_trip(temp_dir: Path) -> None:

+    registry = create_default_registry(temp_dir)

+

+    write_result = await registry.execute(

+        "project_memory_write",

+        memory={"techStack": "Python", "build": "uv run pytest -q"},

+        merge=True,

+    )

+    read_result = await registry.execute("project_memory_read", section="all")

+    directive_result = await registry.execute(

+        "project_memory_add_directive",

+        directive="Use uv, never pip.",

+        priority="high",

+    )

+

+    assert not write_result.is_error

+    assert not read_result.is_error

+    assert not directive_result.is_error

+    memory = json.loads(read_result.output)

+    assert memory["techStack"] == "Python"

+    assert registry.get("project_memory_read").required_permission == PermissionMode.READ_ONLY

+    assert registry.get("project_memory_write").required_permission == PermissionMode.WORKSPACE_WRITE

+

+

+@pytest.mark.asyncio

+async def test_notepad_tools_round_trip(temp_dir: Path) -> None:

+    registry = create_default_registry(temp_dir)

+

+    await registry.execute("notepad_write_priority", content="Keep Loader focused on runtime quality.")

+    await registry.execute("notepad_write_working", content="Audit session resume behavior.")

+    await registry.execute("notepad_write_manual", content="Never delete refs without asking.")

+    read_result = await registry.execute("notepad_read", section="all")

+

+    assert not read_result.is_error

+    assert "Keep Loader focused on runtime quality." in read_result.output

+    assert "Audit session resume behavior." in read_result.output

+    assert "Never delete refs without asking." in read_result.output

+

+

+@pytest.mark.asyncio

+async def test_memory_lifecycle_hook_mirrors_directives_into_notepad(temp_dir: Path) -> None:

+    registry = create_default_registry(temp_dir)

+    policy = build_permission_policy(

+        active_mode=PermissionMode.WORKSPACE_WRITE,

+        workspace_root=temp_dir,

+        tool_requirements=registry.get_tool_requirements(),

+    )

+    executor = ToolExecutor(

+        registry,

+        RuntimeTracer(),

+        policy,

+        hooks=HookManager([MemoryLifecycleHook()]),

+    )

+

+    outcome = await executor.execute_tool_call(

+        ToolCall(

+            id="directive-1",

+            name="project_memory_add_directive",

+            arguments={

+                "directive": "Use uv, never pip.",

+                "priority": "high",

+            },

+        ),

+        source="native",

+        skip_confirmation=True,

+    )

+

+    assert outcome.state == ToolExecutionState.EXECUTED

+    notepad = (temp_dir / ".loader" / "notepad.md").read_text()

+    assert "Remembered directive [high]: Use uv, never pip." in notepad

+

+

+@pytest.mark.asyncio

+async def test_definition_of_done_summary_is_captured_in_project_memory(

+    temp_dir: Path,

+) -> None:

+    target = temp_dir / "memory-proof.txt"

+    backend = ScriptedBackend(

+        completions=[

+            CompletionResponse(

+                content="I'll create the file.",

+                tool_calls=[

+                    ToolCall(

+                        id="write-1",

+                        name="write",

+                        arguments={"file_path": str(target), "content": "memory proof\n"},

+                    )

+                ],

+            ),

+            CompletionResponse(content="The file is in place."),

+        ]

+    )

+

+    await run_scenario(

+        "Create memory-proof.txt in the workspace root.",

+        backend,

+        config=AgentConfig(auto_context=False, stream=False),

+        project_root=temp_dir,

+    )

+

+    memory = json.loads((temp_dir / ".loader" / "project-memory.json").read_text())

+    notes = memory.get("notes", [])

+    assert any(

+        note.get("category") == "definition_of_done"

+        and "Verification:" in note.get("content", "")

+        for note in notes

+    )