Merge pull request #18 from HarnessLab/feature/readme-update-april-2026

Implemented the next parity slice.

PARITY_CHECKLIST.md

@@ -336,6 +336,13 @@ Done:
 - [x] `team_delete`
 - [x] `send_message`
 - [x] `team_messages`
+- [x] `workflow_list`
+- [x] `workflow_get`
+- [x] `workflow_run`
+- [x] `remote_trigger`
+- [x] `worktree_status`
+- [x] `worktree_enter`
+- [x] `worktree_exit`
 
 Missing:
 
@@ -348,11 +355,11 @@ Missing:
 - [ ] Config tool
 - [ ] Terminal capture tool
 - [ ] Browser tool
-- [ ] Workflow tool
-- [ ] Remote trigger tool
+- [x] Workflow tool
+- [x] Remote trigger tool
 - [ ] Sleep / cron tools beyond the current local `sleep` tool
 - [ ] PowerShell tool parity
-- [ ] Worktree enter/exit tools
+- [x] Worktree enter/exit tools
 - [ ] Full `tools.ts` parity
 
 ## 7. Commands And Task Systems
@@ -370,6 +377,9 @@ Done:
 - [x] Local background task management for agent worker sessions
 - [x] Local ask-user runtime with queued answers, history, and slash/CLI inspection flows
 - [x] Local team runtime with persisted teams, messages, and slash/CLI inspection flows
+- [x] Local workflow runtime with manifest discovery, run history, and workflow CLI/slash flows
+- [x] Local remote trigger runtime with create/update/run flows and trigger CLI/slash flows
+- [x] Local managed git worktree runtime with session cwd switching and worktree CLI/slash flows
 
 Missing:
 

README.md

@@ -49,6 +49,9 @@
 | 🆕 | **Ask-User Runtime** | Queued or interactive local ask-user flow with history, slash commands, and agent tool support |
 | 🆕 | **Team Runtime** | Persisted local teams and message history with team/message tools and slash/CLI inspection |
 | 🆕 | **Notebook Edit Tool** | Native `.ipynb` cell editing through the real agent tool registry |
+| 🆕 | **Workflow Runtime** | Manifest-backed local workflows with workflow tools, slash commands, and run history |
+| 🆕 | **Remote Trigger Runtime** | Local remote triggers with create/update/run flows similar to the npm remote trigger surface |
+| 🆕 | **Worktree Runtime** | Managed git worktrees with mid-session cwd switching, slash commands, and CLI flows |
 | 🆕 | **Tokenizer-Aware Context** | Cached tokenizer backends with heuristic fallback for `/context`, `/status`, and compaction |
 | 🆕 | **Daemon Commands** | Local `daemon start/ps/logs/attach/kill` wrapper over background agent sessions |
 | 🆕 | **Background Sessions** | Local `agent-bg`, `agent-ps`, `agent-logs`, `agent-attach`, and `agent-kill` flows |
@@ -92,6 +95,9 @@ Built on the public porting workspace from [instructkr/claw-code](https://github
 | 🙋 **Ask-User Runtime** | Queued answer or interactive user-question flow with history tracking |
 | 👥 **Team Runtime** | Persisted local teams plus message history, handoff notes, and collaboration metadata |
 | 📓 **Notebook Editing** | Native Jupyter notebook cell editing through `notebook_edit` |
+| 🪵 **Worktree Runtime** | Managed git worktrees with `worktree_enter`, `worktree_exit`, and live cwd switching |
+| 🧭 **Workflow Runtime** | Manifest-backed workflows with slash commands, CLI inspection, and recorded runs |
+| ⏰ **Remote Triggers** | Local remote triggers with create/update/run flows and npm-style trigger actions |
 | 🪝 **Hook & Policy Runtime** | Trust reporting, safe env, managed settings, tool blocking, and budget overrides |
 | 🧠 **Context Engine** | Automatic context building with CLAUDE.md discovery, compaction, and snipping |
 | 🔢 **Tokenizer-Aware Accounting** | Model-aware token counting with cached tokenizer backends and fallback heuristics |
@@ -149,6 +155,9 @@ Built on the public porting workspace from [instructkr/claw-code](https://github
 - [x] Local MCP runtime: manifest resources, stdio transport, MCP resources, and MCP tool calls
 - [x] Local task and plan runtimes with plan sync and dependency-aware task execution
 - [x] Notebook edit tool in the real Python tool registry
+- [x] Local workflow runtime with workflow list/get/run tools and CLI/slash flows
+- [x] Local remote trigger runtime with create/update/run flows and CLI/slash inspection
+- [x] Local managed git worktree runtime with live cwd switching and worktree CLI/slash flows
 - [x] Tokenizer-aware context accounting with cached tokenizer backends and heuristic fallback
 - [x] Plugin runtime: manifest discovery, hooks, aliases, virtual tools, tool blocking
 - [x] Plugin lifecycle hooks: resume, persist, delegate phases
@@ -212,6 +221,9 @@ claw-code/
 │   ├── task_runtime.py           # Persistent task runtime and task execution
 │   ├── task.py                   # Task state model and task dataclasses
 │   ├── team_runtime.py           # Local teams, messages, and collaboration metadata
+│   ├── workflow_runtime.py       # Local workflow manifests and recorded workflow runs
+│   ├── remote_trigger_runtime.py # Local remote trigger manifests and trigger run history
+│   ├── worktree_runtime.py       # Managed git worktree sessions and cwd switching
 │   ├── hook_policy.py            # Hook/policy manifests, trust, and safe env handling
 │   ├── tokenizer_runtime.py      # Tokenizer-aware context accounting backends
 │   ├── permissions.py            # Tool permission filtering

TESTING_GUIDE.md

@@ -1450,6 +1450,125 @@ python3 -m src.main agent \
   --show-transcript
 ```
 
+## 18C. Workflow Runtime
+
+### 18C.1 Prepare a workflow manifest
+
+```bash
+cat > ./test_cases/.claw-workflows.json <<'EOF'
+{
+  "workflows": [
+    {
+      "name": "review",
+      "description": "Review the current patch.",
+      "steps": [
+        {"title": "Inspect diff", "detail": "Read {path}"},
+        {"title": "Summarize findings"}
+      ],
+      "prompt": "Review changes under {path}"
+    }
+  ]
+}
+EOF
+```
+
+### 18C.2 CLI inspection and run
+
+```bash
+python3 -m src.main workflow-list --cwd ./test_cases
+python3 -m src.main workflow-get review --cwd ./test_cases
+python3 -m src.main workflow-run review --arguments-json '{"path":"src/"}' --cwd ./test_cases
+```
+
+### 18C.3 Slash commands
+
+```bash
+python3 -m src.main agent "/workflows" --cwd ./test_cases
+python3 -m src.main agent "/workflow review" --cwd ./test_cases
+python3 -m src.main agent "/workflow run review" --cwd ./test_cases
+```
+
+## 18D. Remote Trigger Runtime
+
+### 18D.1 Prepare a trigger manifest
+
+```bash
+cat > ./test_cases/.claw-triggers.json <<'EOF'
+{
+  "triggers": [
+    {
+      "trigger_id": "nightly",
+      "name": "Nightly",
+      "workflow": "review",
+      "schedule": "0 0 * * *",
+      "body": {"depth": "full"}
+    }
+  ]
+}
+EOF
+```
+
+### 18D.2 CLI inspection, create, update, and run
+
+```bash
+python3 -m src.main trigger-list --cwd ./test_cases
+python3 -m src.main trigger-get nightly --cwd ./test_cases
+python3 -m src.main trigger-create --body-json '{"trigger_id":"adhoc","name":"Adhoc","workflow":"review"}' --cwd ./test_cases
+python3 -m src.main trigger-update adhoc --body-json '{"schedule":"manual"}' --cwd ./test_cases
+python3 -m src.main trigger-run nightly --body-json '{"depth":"quick"}' --cwd ./test_cases
+```
+
+### 18D.3 Slash commands
+
+```bash
+python3 -m src.main agent "/triggers" --cwd ./test_cases
+python3 -m src.main agent "/trigger nightly" --cwd ./test_cases
+python3 -m src.main agent "/trigger run nightly" --cwd ./test_cases
+```
+
+## 18E. Worktree Runtime
+
+### 18E.1 Prepare a git workspace
+
+```bash
+cd ./test_cases
+git init
+git config user.email test@example.com
+git config user.name "Test User"
+printf 'hello\n' > README.md
+git add README.md
+git commit -m "init"
+cd ..
+```
+
+### 18E.2 CLI worktree flow
+
+```bash
+python3 -m src.main worktree-status --cwd ./test_cases
+python3 -m src.main worktree-enter preview --cwd ./test_cases
+python3 -m src.main worktree-status --cwd ./test_cases
+python3 -m src.main worktree-exit --action keep --cwd ./test_cases
+```
+
+### 18E.3 Slash commands
+
+```bash
+python3 -m src.main agent "/worktree" --cwd ./test_cases
+python3 -m src.main agent "/worktree enter preview" --cwd ./test_cases
+python3 -m src.main agent "/worktree history" --cwd ./test_cases
+python3 -m src.main agent "/worktree exit remove discard" --cwd ./test_cases
+```
+
+### 18E.4 Real agent cwd switch
+
+```bash
+python3 -m src.main agent \
+  "Create a managed worktree called preview, then write note.txt in the current working directory and summarize where the file was created." \
+  --cwd ./test_cases \
+  --allow-write \
+  --show-transcript
+```
+
 ### 18.2 Create and inspect tasks
 
 ```bash

src/__init__.py

@@ -24,6 +24,7 @@ from .plan_runtime import PlanRuntime, PlanStep
 from .plugin_runtime import PluginRuntime
 from .port_manifest import PortManifest, build_port_manifest
 from .query_engine import QueryEnginePort, TurnResult
+from .remote_trigger_runtime import RemoteTriggerDefinition, RemoteTriggerRunRecord, RemoteTriggerRuntime
 from .runtime import PortRuntime, RuntimeSession
 from .search_runtime import SearchProviderProfile, SearchResult, SearchRuntime, SearchStatusReport
 from .session_store import StoredSession, load_session, save_session
@@ -32,6 +33,8 @@ from .task import PortingTask
 from .task_runtime import TaskRuntime
 from .team_runtime import TeamDefinition, TeamMessage, TeamRuntime
 from .tokenizer_runtime import TokenCounterInfo, clear_token_counter_cache, count_tokens, describe_token_counter
+from .workflow_runtime import WorkflowDefinition, WorkflowRunRecord, WorkflowRuntime
+from .worktree_runtime import WorktreeRuntime, WorktreeSessionState, WorktreeStatusReport
 from .tools import PORTED_TOOLS, build_tool_backlog
 
 __all__ = [
@@ -66,6 +69,9 @@ __all__ = [
     'PortingTask',
     'QueuedUserAnswer',
     'QueryEnginePort',
+    'RemoteTriggerDefinition',
+    'RemoteTriggerRunRecord',
+    'RemoteTriggerRuntime',
     'RuntimeSession',
     'SearchProviderProfile',
     'SearchResult',
@@ -78,6 +84,12 @@ __all__ = [
     'TeamRuntime',
     'TokenCounterInfo',
     'TurnResult',
+    'WorkflowDefinition',
+    'WorkflowRunRecord',
+    'WorkflowRuntime',
+    'WorktreeRuntime',
+    'WorktreeSessionState',
+    'WorktreeStatusReport',
     'PORTED_COMMANDS',
     'PORTED_TOOLS',
     'build_command_backlog',

src/agent_context.py

@@ -17,9 +17,12 @@ from .mcp_runtime import MCPRuntime
 from .plan_runtime import PlanRuntime
 from .plugin_runtime import PluginRuntime
 from .remote_runtime import RemoteRuntime
+from .remote_trigger_runtime import RemoteTriggerRuntime
 from .search_runtime import SearchRuntime
 from .task_runtime import TaskRuntime
 from .team_runtime import TeamRuntime
+from .workflow_runtime import WorkflowRuntime
+from .worktree_runtime import WorktreeRuntime
 from .agent_types import AgentRuntimeConfig
 
 MAX_STATUS_CHARS = 2000
@@ -233,6 +236,12 @@ def _get_user_context_cached(
     remote_runtime = RemoteRuntime.from_workspace(Path(cwd), additional_working_directories)
     if remote_runtime.has_remote_config():
         context['remoteRuntime'] = remote_runtime.render_summary()
+    remote_trigger_runtime = RemoteTriggerRuntime.from_workspace(
+        Path(cwd),
+        additional_working_directories,
+    )
+    if remote_trigger_runtime.has_state():
+        context['remoteTriggerRuntime'] = remote_trigger_runtime.render_summary()
     search_runtime = SearchRuntime.from_workspace(Path(cwd), additional_working_directories)
     if search_runtime.has_search_runtime():
         context['searchRuntime'] = search_runtime.render_summary()
@@ -254,6 +263,12 @@ def _get_user_context_cached(
     team_runtime = TeamRuntime.from_workspace(Path(cwd), additional_working_directories)
     if team_runtime.has_team_state():
         context['teamRuntime'] = team_runtime.render_summary()
+    workflow_runtime = WorkflowRuntime.from_workspace(Path(cwd), additional_working_directories)
+    if workflow_runtime.has_workflows():
+        context['workflowRuntime'] = workflow_runtime.render_summary()
+    worktree_runtime = WorktreeRuntime.from_workspace(Path(cwd))
+    if worktree_runtime.repo_root is not None or worktree_runtime.has_state():
+        context['worktreeRuntime'] = worktree_runtime.render_summary()
     return context
 
 

src/agent_runtime.py

@@ -47,10 +47,13 @@ from .openai_compat import OpenAICompatClient, OpenAICompatError
 from .plan_runtime import PlanRuntime
 from .plugin_runtime import PluginRuntime
 from .remote_runtime import RemoteRuntime
+from .remote_trigger_runtime import RemoteTriggerRuntime
 from .search_runtime import SearchRuntime
 from .task_runtime import TaskRuntime
 from .team_runtime import TeamRuntime
 from .tokenizer_runtime import describe_token_counter
+from .workflow_runtime import WorkflowRuntime
+from .worktree_runtime import WorktreeRuntime
 from .session_store import (
     StoredAgentSession,
     load_agent_session,
@@ -84,6 +87,7 @@ class LocalCodingAgent:
     hook_policy_runtime: HookPolicyRuntime | None = None
     mcp_runtime: MCPRuntime | None = None
     remote_runtime: RemoteRuntime | None = None
+    remote_trigger_runtime: RemoteTriggerRuntime | None = None
     search_runtime: SearchRuntime | None = None
     account_runtime: AccountRuntime | None = None
     ask_user_runtime: AskUserRuntime | None = None
@@ -91,6 +95,8 @@ class LocalCodingAgent:
     plan_runtime: PlanRuntime | None = None
     task_runtime: TaskRuntime | None = None
     team_runtime: TeamRuntime | None = None
+    workflow_runtime: WorkflowRuntime | None = None
+    worktree_runtime: WorktreeRuntime | None = None
     last_session: AgentSessionState | None = field(default=None, init=False, repr=False)
     last_run_result: AgentRunResult | None = field(default=None, init=False, repr=False)
     active_session_id: str | None = field(default=None, init=False, repr=False)
@@ -123,6 +129,11 @@ class LocalCodingAgent:
                 self.runtime_config.cwd,
                 tuple(str(path) for path in self.runtime_config.additional_working_directories),
             )
+        if self.remote_trigger_runtime is None:
+            self.remote_trigger_runtime = RemoteTriggerRuntime.from_workspace(
+                self.runtime_config.cwd,
+                tuple(str(path) for path in self.runtime_config.additional_working_directories),
+            )
         if self.search_runtime is None:
             self.search_runtime = SearchRuntime.from_workspace(
                 self.runtime_config.cwd,
@@ -149,6 +160,13 @@ class LocalCodingAgent:
                 self.runtime_config.cwd,
                 tuple(str(path) for path in self.runtime_config.additional_working_directories),
             )
+        if self.workflow_runtime is None:
+            self.workflow_runtime = WorkflowRuntime.from_workspace(
+                self.runtime_config.cwd,
+                tuple(str(path) for path in self.runtime_config.additional_working_directories),
+            )
+        if self.worktree_runtime is None:
+            self.worktree_runtime = WorktreeRuntime.from_workspace(self.runtime_config.cwd)
         self.runtime_config = self._apply_hook_policy_budget_overrides(self.runtime_config)
         registry = dict(self.tool_registry)
         plugin_tools = self.plugin_runtime.register_tool_aliases(registry)
@@ -173,9 +191,12 @@ class LocalCodingAgent:
             config_runtime=self.config_runtime,
             mcp_runtime=self.mcp_runtime,
             remote_runtime=self.remote_runtime,
+            remote_trigger_runtime=self.remote_trigger_runtime,
             plan_runtime=self.plan_runtime,
             task_runtime=self.task_runtime,
             team_runtime=self.team_runtime,
+            workflow_runtime=self.workflow_runtime,
+            worktree_runtime=self.worktree_runtime,
         )
 
     def set_model(self, model: str) -> None:
@@ -3021,6 +3042,45 @@ class LocalCodingAgent:
         clear_context_caches()
         return '\n'.join(['# Remote', '', report.as_text()])
 
+    def render_worktree_report(self) -> str:
+        if self.worktree_runtime is None:
+            return '# Worktree\n\nNo local worktree runtime is available.'
+        return '\n'.join(['# Worktree', '', self.worktree_runtime.render_summary()])
+
+    def render_worktree_enter_report(self, name: str | None = None) -> str:
+        if self.worktree_runtime is None:
+            return '# Worktree\n\nNo local worktree runtime is available.'
+        try:
+            report = self.worktree_runtime.enter(name=name)
+        except (RuntimeError, ValueError) as exc:
+            return f'# Worktree\n\n{exc}'
+        self._apply_runtime_cwd_update(Path(report.worktree_path or self.runtime_config.cwd))
+        return '\n'.join(['# Worktree', '', report.as_text()])
+
+    def render_worktree_exit_report(
+        self,
+        *,
+        action: str = 'keep',
+        discard_changes: bool = False,
+    ) -> str:
+        if self.worktree_runtime is None:
+            return '# Worktree\n\nNo local worktree runtime is available.'
+        try:
+            report = self.worktree_runtime.exit(
+                action=action,
+                discard_changes=discard_changes,
+            )
+        except (RuntimeError, ValueError) as exc:
+            return f'# Worktree\n\n{exc}'
+        target_cwd = report.original_cwd or report.current_cwd or str(self.runtime_config.cwd)
+        self._apply_runtime_cwd_update(Path(target_cwd))
+        return '\n'.join(['# Worktree', '', report.as_text()])
+
+    def render_worktree_history_report(self) -> str:
+        if self.worktree_runtime is None:
+            return '# Worktree History\n\nNo local worktree runtime is available.'
+        return self.worktree_runtime.render_history()
+
     def render_mcp_resources_report(self, query: str | None = None) -> str:
         if self.mcp_runtime is None:
             return '# MCP Resources\n\nNo local MCP manifests, servers, or resources discovered.'
@@ -3110,6 +3170,81 @@ class LocalCodingAgent:
         except KeyError:
             return f'# Team Messages\n\nUnknown team: {team_name}'
 
+    def render_workflows_report(self, query: str | None = None) -> str:
+        if self.workflow_runtime is None or not self.workflow_runtime.has_workflows():
+            return '# Workflows\n\nNo local workflow runtime is available.'
+        return self.workflow_runtime.render_workflows_index(query=query)
+
+    def render_workflow_report(self, workflow_name: str) -> str:
+        if self.workflow_runtime is None or not self.workflow_runtime.has_workflows():
+            return '# Workflow\n\nNo local workflow runtime is available.'
+        try:
+            return self.workflow_runtime.render_workflow(workflow_name)
+        except KeyError:
+            return f'# Workflow\n\nUnknown workflow: {workflow_name}'
+
+    def render_workflow_run_report(
+        self,
+        workflow_name: str,
+        *,
+        arguments: dict[str, Any] | None = None,
+    ) -> str:
+        if self.workflow_runtime is None or not self.workflow_runtime.has_workflows():
+            return '# Workflow Run\n\nNo local workflow runtime is available.'
+        try:
+            return self.workflow_runtime.render_run_report(
+                workflow_name,
+                arguments=arguments,
+            )
+        except KeyError:
+            return f'# Workflow Run\n\nUnknown workflow: {workflow_name}'
+
+    def render_remote_triggers_report(self, query: str | None = None) -> str:
+        if self.remote_trigger_runtime is None or not self.remote_trigger_runtime.has_state():
+            return '# Remote Triggers\n\nNo local remote trigger runtime is available.'
+        return self.remote_trigger_runtime.render_trigger_index(query=query)
+
+    def render_remote_trigger_report(self, trigger_id: str) -> str:
+        if self.remote_trigger_runtime is None or not self.remote_trigger_runtime.has_state():
+            return '# Remote Trigger\n\nNo local remote trigger runtime is available.'
+        try:
+            return self.remote_trigger_runtime.render_trigger(trigger_id)
+        except KeyError:
+            return f'# Remote Trigger\n\nUnknown remote trigger: {trigger_id}'
+
+    def render_remote_trigger_action_report(
+        self,
+        action: str,
+        *,
+        trigger_id: str | None = None,
+        body: dict[str, Any] | None = None,
+    ) -> str:
+        if self.remote_trigger_runtime is None:
+            return '# Remote Trigger\n\nNo local remote trigger runtime is available.'
+        normalized = action.strip().lower()
+        try:
+            if normalized == 'list':
+                return self.remote_trigger_runtime.render_trigger_index()
+            if normalized == 'get':
+                if not trigger_id:
+                    return '# Remote Trigger\n\ntrigger_id is required for get'
+                return self.remote_trigger_runtime.render_trigger(trigger_id)
+            if normalized == 'create':
+                created = self.remote_trigger_runtime.create_trigger(body or {})
+                return self.remote_trigger_runtime.render_trigger(created.trigger_id)
+            if normalized == 'update':
+                if not trigger_id:
+                    return '# Remote Trigger\n\ntrigger_id is required for update'
+                updated = self.remote_trigger_runtime.update_trigger(trigger_id, body or {})
+                return self.remote_trigger_runtime.render_trigger(updated.trigger_id)
+            if normalized == 'run':
+                if not trigger_id:
+                    return '# Remote Trigger Run\n\ntrigger_id is required for run'
+                return self.remote_trigger_runtime.render_run_report(trigger_id, body=body)
+        except (KeyError, TypeError, ValueError) as exc:
+            return f'# Remote Trigger\n\n{exc}'
+        return '# Remote Trigger\n\naction must be one of list, get, create, update, or run'
+
     def render_hook_policy_report(self) -> str:
         if self.hook_policy_runtime is None:
             return '# Hook Policy\n\nNo local hook or policy manifests discovered.'
@@ -3235,6 +3370,9 @@ class LocalCodingAgent:
     ) -> None:
         if not tool_result.ok:
             return
+        cwd_update = tool_result.metadata.get('cwd_update')
+        if isinstance(cwd_update, str) and cwd_update:
+            self._apply_runtime_cwd_update(Path(cwd_update))
         refresh_tool_names = {
             'update_plan',
             'plan_clear',
@@ -3255,6 +3393,10 @@ class LocalCodingAgent:
             'team_create',
             'team_delete',
             'send_message',
+            'workflow_run',
+            'remote_trigger',
+            'worktree_enter',
+            'worktree_exit',
         }
         if tool_name not in refresh_tool_names:
             return
@@ -3267,6 +3409,11 @@ class LocalCodingAgent:
                 self.runtime_config.cwd,
                 additional_working_directories=additional_dirs,
             )
+        if tool_name == 'remote_trigger':
+            self.remote_trigger_runtime = RemoteTriggerRuntime.from_workspace(
+                self.runtime_config.cwd,
+                additional_working_directories=additional_dirs,
+            )
         if tool_name.startswith('search_'):
             self.search_runtime = SearchRuntime.from_workspace(
                 self.runtime_config.cwd,
@@ -3293,6 +3440,13 @@ class LocalCodingAgent:
                 self.runtime_config.cwd,
                 additional_working_directories=additional_dirs,
             )
+        if tool_name.startswith('workflow_'):
+            self.workflow_runtime = WorkflowRuntime.from_workspace(
+                self.runtime_config.cwd,
+                additional_working_directories=additional_dirs,
+            )
+        if tool_name.startswith('worktree_'):
+            self.worktree_runtime = WorktreeRuntime.from_workspace(self.runtime_config.cwd)
         self.tool_context = replace(
             self.tool_context,
             tool_registry=self.tool_registry,
@@ -3301,9 +3455,97 @@ class LocalCodingAgent:
             ask_user_runtime=self.ask_user_runtime,
             config_runtime=self.config_runtime,
             remote_runtime=self.remote_runtime,
+            remote_trigger_runtime=self.remote_trigger_runtime,
             plan_runtime=self.plan_runtime,
             task_runtime=self.task_runtime,
             team_runtime=self.team_runtime,
+            workflow_runtime=self.workflow_runtime,
+            worktree_runtime=self.worktree_runtime,
+        )
+
+    def _apply_runtime_cwd_update(self, new_cwd: Path) -> None:
+        resolved_cwd = new_cwd.resolve()
+        if resolved_cwd == self.runtime_config.cwd.resolve():
+            return
+        self.runtime_config = replace(self.runtime_config, cwd=resolved_cwd)
+        clear_context_caches()
+        additional_dirs = tuple(
+            str(path) for path in self.runtime_config.additional_working_directories
+        )
+        self.plugin_runtime = PluginRuntime.from_workspace(
+            self.runtime_config.cwd,
+            additional_dirs,
+        )
+        self.hook_policy_runtime = HookPolicyRuntime.from_workspace(
+            self.runtime_config.cwd,
+            additional_dirs,
+        )
+        self.mcp_runtime = MCPRuntime.from_workspace(
+            self.runtime_config.cwd,
+            additional_dirs,
+        )
+        self.remote_runtime = RemoteRuntime.from_workspace(
+            self.runtime_config.cwd,
+            additional_dirs,
+        )
+        self.remote_trigger_runtime = RemoteTriggerRuntime.from_workspace(
+            self.runtime_config.cwd,
+            additional_dirs,
+        )
+        self.search_runtime = SearchRuntime.from_workspace(
+            self.runtime_config.cwd,
+            additional_dirs,
+        )
+        self.account_runtime = AccountRuntime.from_workspace(
+            self.runtime_config.cwd,
+            additional_dirs,
+        )
+        self.ask_user_runtime = AskUserRuntime.from_workspace(
+            self.runtime_config.cwd,
+            additional_dirs,
+        )
+        self.config_runtime = ConfigRuntime.from_workspace(self.runtime_config.cwd)
+        self.task_runtime = TaskRuntime.from_workspace(self.runtime_config.cwd)
+        self.plan_runtime = PlanRuntime.from_workspace(self.runtime_config.cwd)
+        self.team_runtime = TeamRuntime.from_workspace(
+            self.runtime_config.cwd,
+            additional_dirs,
+        )
+        self.workflow_runtime = WorkflowRuntime.from_workspace(
+            self.runtime_config.cwd,
+            additional_dirs,
+        )
+        self.worktree_runtime = WorktreeRuntime.from_workspace(self.runtime_config.cwd)
+        self.runtime_config = self._apply_hook_policy_budget_overrides(self.runtime_config)
+        registry = dict(default_tool_registry())
+        if self.plugin_runtime is not None:
+            alias_tools = self.plugin_runtime.register_tool_aliases(registry)
+            if alias_tools:
+                registry = {**registry, **alias_tools}
+            virtual_tools = self.plugin_runtime.register_virtual_tools(registry)
+            if virtual_tools:
+                registry = {**registry, **virtual_tools}
+        self.tool_registry = registry
+        self.tool_context = build_tool_context(
+            self.runtime_config,
+            tool_registry=self.tool_registry,
+            extra_env=(
+                self.hook_policy_runtime.safe_env()
+                if self.hook_policy_runtime is not None
+                else None
+            ),
+            search_runtime=self.search_runtime,
+            account_runtime=self.account_runtime,
+            ask_user_runtime=self.ask_user_runtime,
+            config_runtime=self.config_runtime,
+            mcp_runtime=self.mcp_runtime,
+            remote_runtime=self.remote_runtime,
+            remote_trigger_runtime=self.remote_trigger_runtime,
+            plan_runtime=self.plan_runtime,
+            task_runtime=self.task_runtime,
+            team_runtime=self.team_runtime,
+            workflow_runtime=self.workflow_runtime,
+            worktree_runtime=self.worktree_runtime,
         )
 
     def _apply_plugin_before_prompt_hooks(self, prompt: str) -> str:

src/agent_slash_commands.py

@@ -124,6 +124,11 @@ def get_slash_command_specs() -> tuple[SlashCommandSpec, ...]:
             description='Show local remote runtime status or activate a remote target/profile.',
             handler=_handle_remote,
         ),
+        SlashCommandSpec(
+            names=('worktree',),
+            description='Show managed git worktree status or enter/exit the current managed worktree session.',
+            handler=_handle_worktree,
+        ),
         SlashCommandSpec(
             names=('account',),
             description='Show local account runtime status or configured account profiles.',
@@ -194,6 +199,26 @@ def get_slash_command_specs() -> tuple[SlashCommandSpec, ...]:
             description='Show the local runtime task list, optionally filtered by status.',
             handler=_handle_tasks,
         ),
+        SlashCommandSpec(
+            names=('workflows',),
+            description='List local workflows discovered from workflow manifests.',
+            handler=_handle_workflows,
+        ),
+        SlashCommandSpec(
+            names=('workflow',),
+            description='Show or run one local workflow by name.',
+            handler=_handle_workflow,
+        ),
+        SlashCommandSpec(
+            names=('triggers',),
+            description='List local remote triggers discovered from remote trigger manifests.',
+            handler=_handle_triggers,
+        ),
+        SlashCommandSpec(
+            names=('trigger',),
+            description='Show or run one local remote trigger by id.',
+            handler=_handle_trigger,
+        ),
         SlashCommandSpec(
             names=('teams',),
             description='List the locally configured collaboration teams.',
@@ -349,6 +374,32 @@ def _handle_remote(agent: 'LocalCodingAgent', args: str, input_text: str) -> Sla
     return _local_result(input_text, agent.render_remote_report(target))
 
 
+def _handle_worktree(agent: 'LocalCodingAgent', args: str, input_text: str) -> SlashCommandResult:
+    command = args.strip()
+    if not command:
+        return _local_result(input_text, agent.render_worktree_report())
+    if command == 'history':
+        return _local_result(input_text, agent.render_worktree_history_report())
+    if command.startswith('enter'):
+        name = command.split(' ', 1)[1].strip() if ' ' in command else None
+        return _local_result(input_text, agent.render_worktree_enter_report(name or None))
+    if command.startswith('exit'):
+        parts = command.split()
+        action = parts[1] if len(parts) > 1 else 'keep'
+        discard_changes = any(part in {'discard', 'discard_changes=true'} for part in parts[2:])
+        return _local_result(
+            input_text,
+            agent.render_worktree_exit_report(
+                action=action,
+                discard_changes=discard_changes,
+            ),
+        )
+    return _local_result(
+        input_text,
+        'Usage: /worktree [history|enter <name>|exit <keep|remove> [discard]]',
+    )
+
+
 def _handle_account(agent: 'LocalCodingAgent', args: str, input_text: str) -> SlashCommandResult:
     command = args.strip()
     if not command:
@@ -458,6 +509,43 @@ def _handle_tasks(agent: 'LocalCodingAgent', args: str, input_text: str) -> Slas
     return _local_result(input_text, agent.render_tasks_report(status))
 
 
+def _handle_workflows(agent: 'LocalCodingAgent', args: str, input_text: str) -> SlashCommandResult:
+    query = args or None
+    return _local_result(input_text, agent.render_workflows_report(query))
+
+
+def _handle_workflow(agent: 'LocalCodingAgent', args: str, input_text: str) -> SlashCommandResult:
+    command = args.strip()
+    if not command:
+        return _local_result(input_text, 'Usage: /workflow <name> | /workflow run <name>')
+    if command.startswith('run '):
+        workflow_name = command.split(' ', 1)[1].strip()
+        if not workflow_name:
+            return _local_result(input_text, 'Usage: /workflow run <name>')
+        return _local_result(input_text, agent.render_workflow_run_report(workflow_name))
+    return _local_result(input_text, agent.render_workflow_report(command))
+
+
+def _handle_triggers(agent: 'LocalCodingAgent', args: str, input_text: str) -> SlashCommandResult:
+    query = args or None
+    return _local_result(input_text, agent.render_remote_triggers_report(query))
+
+
+def _handle_trigger(agent: 'LocalCodingAgent', args: str, input_text: str) -> SlashCommandResult:
+    command = args.strip()
+    if not command:
+        return _local_result(input_text, 'Usage: /trigger <id> | /trigger run <id>')
+    if command.startswith('run '):
+        trigger_id = command.split(' ', 1)[1].strip()
+        if not trigger_id:
+            return _local_result(input_text, 'Usage: /trigger run <id>')
+        return _local_result(
+            input_text,
+            agent.render_remote_trigger_action_report('run', trigger_id=trigger_id),
+        )
+    return _local_result(input_text, agent.render_remote_trigger_report(command))
+
+
 def _handle_teams(agent: 'LocalCodingAgent', args: str, input_text: str) -> SlashCommandResult:
     query = args or None
     return _local_result(input_text, agent.render_teams_report(query))

src/agent_tools.py

@@ -23,9 +23,12 @@ if TYPE_CHECKING:
     from .mcp_runtime import MCPRuntime
     from .plan_runtime import PlanRuntime
     from .remote_runtime import RemoteRuntime
+    from .remote_trigger_runtime import RemoteTriggerRuntime
     from .search_runtime import SearchRuntime
     from .task_runtime import TaskRuntime
     from .team_runtime import TeamRuntime
+    from .workflow_runtime import WorkflowRuntime
+    from .worktree_runtime import WorktreeRuntime
 
 
 class ToolPermissionError(RuntimeError):
@@ -50,9 +53,12 @@ class ToolExecutionContext:
     config_runtime: 'ConfigRuntime | None' = None
     mcp_runtime: 'MCPRuntime | None' = None
     remote_runtime: 'RemoteRuntime | None' = None
+    remote_trigger_runtime: 'RemoteTriggerRuntime | None' = None
     plan_runtime: 'PlanRuntime | None' = None
     task_runtime: 'TaskRuntime | None' = None
     team_runtime: 'TeamRuntime | None' = None
+    workflow_runtime: 'WorkflowRuntime | None' = None
+    worktree_runtime: 'WorktreeRuntime | None' = None
 
 
 ToolHandler = Callable[
@@ -122,9 +128,12 @@ def build_tool_context(
     config_runtime: 'ConfigRuntime | None' = None,
     mcp_runtime: 'MCPRuntime | None' = None,
     remote_runtime: 'RemoteRuntime | None' = None,
+    remote_trigger_runtime: 'RemoteTriggerRuntime | None' = None,
     plan_runtime: 'PlanRuntime | None' = None,
     task_runtime: 'TaskRuntime | None' = None,
     team_runtime: 'TeamRuntime | None' = None,
+    workflow_runtime: 'WorkflowRuntime | None' = None,
+    worktree_runtime: 'WorktreeRuntime | None' = None,
 ) -> ToolExecutionContext:
     return ToolExecutionContext(
         root=config.cwd.resolve(),
@@ -139,9 +148,12 @@ def build_tool_context(
         config_runtime=config_runtime,
         mcp_runtime=mcp_runtime,
         remote_runtime=remote_runtime,
+        remote_trigger_runtime=remote_trigger_runtime,
         plan_runtime=plan_runtime,
         task_runtime=task_runtime,
         team_runtime=team_runtime,
+        workflow_runtime=workflow_runtime,
+        worktree_runtime=worktree_runtime,
     )
 
 
@@ -610,6 +622,91 @@ def default_tool_registry() -> dict[str, AgentTool]:
             },
             handler=_remote_disconnect,
         ),
+        AgentTool(
+            name='worktree_status',
+            description='Show the current managed git worktree session status.',
+            parameters={
+                'type': 'object',
+                'properties': {},
+            },
+            handler=_worktree_status,
+        ),
+        AgentTool(
+            name='worktree_enter',
+            description='Create an isolated git worktree and switch the current agent session into it.',
+            parameters={
+                'type': 'object',
+                'properties': {
+                    'name': {'type': 'string'},
+                },
+            },
+            handler=_worktree_enter,
+        ),
+        AgentTool(
+            name='worktree_exit',
+            description='Leave the active managed worktree session and optionally remove the worktree.',
+            parameters={
+                'type': 'object',
+                'properties': {
+                    'action': {'type': 'string'},
+                    'discard_changes': {'type': 'boolean'},
+                },
+            },
+            handler=_worktree_exit,
+        ),
+        AgentTool(
+            name='workflow_list',
+            description='List local workflow definitions discovered from workspace workflow manifests.',
+            parameters={
+                'type': 'object',
+                'properties': {
+                    'query': {'type': 'string'},
+                    'max_workflows': {'type': 'integer', 'minimum': 1, 'maximum': 200},
+                },
+            },
+            handler=_workflow_list,
+        ),
+        AgentTool(
+            name='workflow_get',
+            description='Show one local workflow definition by name.',
+            parameters={
+                'type': 'object',
+                'properties': {
+                    'workflow_name': {'type': 'string'},
+                },
+                'required': ['workflow_name'],
+            },
+            handler=_workflow_get,
+        ),
+        AgentTool(
+            name='workflow_run',
+            description='Record and render a local workflow execution request from a workflow manifest.',
+            parameters={
+                'type': 'object',
+                'properties': {
+                    'workflow_name': {'type': 'string'},
+                    'arguments': {'type': 'object'},
+                },
+                'required': ['workflow_name'],
+            },
+            handler=_workflow_run,
+        ),
+        AgentTool(
+            name='remote_trigger',
+            description='List, inspect, create, update, or run local remote triggers similar to the npm remote trigger tool.',
+            parameters={
+                'type': 'object',
+                'properties': {
+                    'action': {'type': 'string'},
+                    'trigger_id': {'type': 'string'},
+                    'body': {'type': 'object'},
+                    'query': {'type': 'string'},
+                    'max_triggers': {'type': 'integer', 'minimum': 1, 'maximum': 200},
+                },
+                'required': ['action'],
+            },
+            handler=_remote_trigger,
+        ),
         AgentTool(
             name='plan_get',
             description='Show the current local runtime plan.',
@@ -1843,6 +1940,213 @@ def _remote_disconnect(
     )
 
 
+def _worktree_status(
+    arguments: dict[str, Any],
+    context: ToolExecutionContext,
+) -> str:
+    _ = arguments
+    runtime = _require_worktree_runtime(context)
+    return '\n'.join(['# Worktree', '', runtime.render_summary()])
+
+
+def _worktree_enter(
+    arguments: dict[str, Any],
+    context: ToolExecutionContext,
+) -> tuple[str, dict[str, Any]]:
+    runtime = _require_worktree_runtime(context)
+    name = arguments.get('name')
+    if name is not None and not isinstance(name, str):
+        raise ToolExecutionError('name must be a string')
+    try:
+        report = runtime.enter(name=name)
+    except (RuntimeError, ValueError) as exc:
+        raise ToolExecutionError(str(exc)) from exc
+    return (
+        report.as_text(),
+        {
+            'action': 'worktree_enter',
+            'cwd_update': report.worktree_path,
+            'repo_root': report.repo_root,
+            'worktree_path': report.worktree_path,
+            'worktree_branch': report.worktree_branch,
+            'session_name': report.session_name,
+            'before_cwd': report.original_cwd,
+            'after_cwd': report.worktree_path,
+            'path': report.worktree_path,
+        },
+    )
+
+
+def _worktree_exit(
+    arguments: dict[str, Any],
+    context: ToolExecutionContext,
+) -> tuple[str, dict[str, Any]]:
+    runtime = _require_worktree_runtime(context)
+    action = arguments.get('action', 'keep')
+    discard_changes = arguments.get('discard_changes', False)
+    if not isinstance(action, str):
+        raise ToolExecutionError('action must be a string')
+    if not isinstance(discard_changes, bool):
+        raise ToolExecutionError('discard_changes must be a boolean')
+    try:
+        report = runtime.exit(
+            action=action,
+            discard_changes=discard_changes,
+        )
+    except (RuntimeError, ValueError) as exc:
+        raise ToolExecutionError(str(exc)) from exc
+    return (
+        report.as_text(),
+        {
+            'action': 'worktree_exit',
+            'cwd_update': report.original_cwd or report.current_cwd,
+            'repo_root': report.repo_root,
+            'worktree_path': report.worktree_path,
+            'worktree_branch': report.worktree_branch,
+            'session_name': report.session_name,
+            'before_cwd': report.worktree_path,
+            'after_cwd': report.original_cwd or report.current_cwd,
+            'exit_action': report.metadata.get('action'),
+            'discard_changes': report.metadata.get('discard_changes'),
+            'path': report.worktree_path,
+        },
+    )
+
+
+def _workflow_list(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
+    runtime = _require_workflow_runtime(context)
+    query = arguments.get('query')
+    if query is not None and not isinstance(query, str):
+        raise ToolExecutionError('query must be a string')
+    max_workflows = _coerce_int(arguments, 'max_workflows', 50)
+    workflows = runtime.list_workflows(query=query, limit=max_workflows)
+    lines = ['# Workflows', '']
+    if not workflows:
+        lines.append('No local workflows discovered.')
+        return '\n'.join(lines)
+    for workflow in workflows:
+        description = workflow.description or 'No description.'
+        lines.append(f'- {workflow.name} ; steps={len(workflow.steps)} ; {description}')
+    return '\n'.join(lines)
+
+
+def _workflow_get(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
+    runtime = _require_workflow_runtime(context)
+    workflow_name = _require_string(arguments, 'workflow_name')
+    try:
+        return runtime.render_workflow(workflow_name)
+    except KeyError as exc:
+        raise ToolExecutionError(f'Unknown workflow: {workflow_name}') from exc
+
+
+def _workflow_run(
+    arguments: dict[str, Any],
+    context: ToolExecutionContext,
+) -> tuple[str, dict[str, Any]]:
+    runtime = _require_workflow_runtime(context)
+    workflow_name = _require_string(arguments, 'workflow_name')
+    raw_arguments = arguments.get('arguments', {})
+    if raw_arguments is None:
+        raw_arguments = {}
+    if not isinstance(raw_arguments, dict):
+        raise ToolExecutionError('arguments must be an object')
+    try:
+        rendered = runtime.render_run_report(workflow_name, arguments=raw_arguments)
+    except KeyError as exc:
+        raise ToolExecutionError(f'Unknown workflow: {workflow_name}') from exc
+    return (
+        rendered,
+        {
+            'action': 'workflow_run',
+            'workflow_name': workflow_name,
+            'argument_keys': sorted(str(key) for key in raw_arguments.keys()),
+        },
+    )
+
+
+def _remote_trigger(
+    arguments: dict[str, Any],
+    context: ToolExecutionContext,
+) -> tuple[str, dict[str, Any]]:
+    runtime = _require_remote_trigger_runtime(context)
+    action = _require_string(arguments, 'action').strip().lower()
+    if action == 'list':
+        query = arguments.get('query')
+        if query is not None and not isinstance(query, str):
+            raise ToolExecutionError('query must be a string')
+        max_triggers = _coerce_int(arguments, 'max_triggers', 50)
+        rendered = runtime.render_trigger_index(query=query)
+        return (
+            rendered,
+            {
+                'action': 'remote_trigger',
+                'remote_trigger_action': action,
+                'listed_triggers': len(runtime.list_triggers(query=query, limit=max_triggers)),
+            },
+        )
+    if action == 'get':
+        trigger_id = _require_string(arguments, 'trigger_id')
+        try:
+            rendered = runtime.render_trigger(trigger_id)
+        except KeyError as exc:
+            raise ToolExecutionError(f'Unknown remote trigger: {trigger_id}') from exc
+        return (
+            rendered,
+            {
+                'action': 'remote_trigger',
+                'remote_trigger_action': action,
+                'trigger_id': trigger_id,
+            },
+        )
+    body = arguments.get('body', {})
+    if body is None:
+        body = {}
+    if not isinstance(body, dict):
+        raise ToolExecutionError('body must be an object')
+    if action == 'create':
+        try:
+            trigger = runtime.create_trigger(body)
+        except (KeyError, TypeError, ValueError) as exc:
+            raise ToolExecutionError(str(exc)) from exc
+        return (
+            runtime.render_trigger(trigger.trigger_id),
+            {
+                'action': 'remote_trigger',
+                'remote_trigger_action': action,
+                'trigger_id': trigger.trigger_id,
+            },
+        )
+    trigger_id = _require_string(arguments, 'trigger_id')
+    if action == 'update':
+        try:
+            trigger = runtime.update_trigger(trigger_id, body)
+        except (KeyError, TypeError, ValueError) as exc:
+            raise ToolExecutionError(str(exc)) from exc
+        return (
+            runtime.render_trigger(trigger.trigger_id),
+            {
+                'action': 'remote_trigger',
+                'remote_trigger_action': action,
+                'trigger_id': trigger.trigger_id,
+            },
+        )
+    if action == 'run':
+        try:
+            rendered = runtime.render_run_report(trigger_id, body=body)
+        except KeyError as exc:
+            raise ToolExecutionError(f'Unknown remote trigger: {trigger_id}') from exc
+        return (
+            rendered,
+            {
+                'action': 'remote_trigger',
+                'remote_trigger_action': action,
+                'trigger_id': trigger_id,
+                'body_keys': sorted(str(key) for key in body.keys()),
+            },
+        )
+    raise ToolExecutionError('action must be one of list, get, create, update, or run')
+
+
 def _team_list(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
     runtime = _require_team_runtime(context)
     query = arguments.get('query')
@@ -2472,6 +2776,12 @@ def _require_remote_runtime(context: ToolExecutionContext):
     return context.remote_runtime
 
 
+def _require_remote_trigger_runtime(context: ToolExecutionContext):
+    if context.remote_trigger_runtime is None:
+        raise ToolExecutionError('Local remote trigger runtime is not available.')
+    return context.remote_trigger_runtime
+
+
 def _require_plan_runtime(context: ToolExecutionContext):
     if context.plan_runtime is None:
         raise ToolExecutionError('Local plan runtime is not available.')
@@ -2490,6 +2800,18 @@ def _require_team_runtime(context: ToolExecutionContext):
     return context.team_runtime
 
 
+def _require_workflow_runtime(context: ToolExecutionContext):
+    if context.workflow_runtime is None or not context.workflow_runtime.has_workflows():
+        raise ToolExecutionError('Local workflow runtime is not available.')
+    return context.workflow_runtime
+
+
+def _require_worktree_runtime(context: ToolExecutionContext):
+    if context.worktree_runtime is None:
+        raise ToolExecutionError('Local worktree runtime is not available.')
+    return context.worktree_runtime
+
+
 def _task_mutation_metadata(
     *,
     action: str,

src/main.py

@@ -37,8 +37,12 @@ from .remote_runtime import (
     run_ssh_mode,
     run_teleport_mode,
 )
+from .remote_trigger_runtime import RemoteTriggerRuntime
 from .search_runtime import SearchRuntime
 from .team_runtime import TeamRuntime
+from .task_runtime import TaskRuntime
+from .workflow_runtime import WorkflowRuntime
+from .worktree_runtime import WorktreeRuntime
 from .runtime import PortRuntime
 from .session_store import (
     StoredAgentSession,
@@ -603,6 +607,15 @@ def build_parser() -> argparse.ArgumentParser:
     remote_profiles_parser.add_argument('--query')
     remote_disconnect_parser = subparsers.add_parser('remote-disconnect', help='disconnect the active local remote target')
     remote_disconnect_parser.add_argument('--cwd', default='.')
+    worktree_status_parser = subparsers.add_parser('worktree-status', help='show local managed git worktree status')
+    worktree_status_parser.add_argument('--cwd', default='.')
+    worktree_enter_parser = subparsers.add_parser('worktree-enter', help='create and enter a managed git worktree')
+    worktree_enter_parser.add_argument('name', nargs='?')
+    worktree_enter_parser.add_argument('--cwd', default='.')
+    worktree_exit_parser = subparsers.add_parser('worktree-exit', help='exit the active managed git worktree')
+    worktree_exit_parser.add_argument('--action', default='keep')
+    worktree_exit_parser.add_argument('--discard-changes', action='store_true')
+    worktree_exit_parser.add_argument('--cwd', default='.')
     account_status_parser = subparsers.add_parser('account-status', help='show local account runtime status')
     account_status_parser.add_argument('--cwd', default='.')
     account_profiles_parser = subparsers.add_parser('account-profiles', help='list configured local account profiles')
@@ -667,6 +680,33 @@ def build_parser() -> argparse.ArgumentParser:
     config_set_parser.add_argument('value_json')
     config_set_parser.add_argument('--source', default='local')
     config_set_parser.add_argument('--cwd', default='.')
+    workflow_list_parser = subparsers.add_parser('workflow-list', help='list local workflow definitions')
+    workflow_list_parser.add_argument('--cwd', default='.')
+    workflow_list_parser.add_argument('--query')
+    workflow_get_parser = subparsers.add_parser('workflow-get', help='show one local workflow definition')
+    workflow_get_parser.add_argument('workflow_name')
+    workflow_get_parser.add_argument('--cwd', default='.')
+    workflow_run_parser = subparsers.add_parser('workflow-run', help='record and render a local workflow run')
+    workflow_run_parser.add_argument('workflow_name')
+    workflow_run_parser.add_argument('--arguments-json', default='{}')
+    workflow_run_parser.add_argument('--cwd', default='.')
+    trigger_list_parser = subparsers.add_parser('trigger-list', help='list local remote triggers')
+    trigger_list_parser.add_argument('--cwd', default='.')
+    trigger_list_parser.add_argument('--query')
+    trigger_get_parser = subparsers.add_parser('trigger-get', help='show one local remote trigger')
+    trigger_get_parser.add_argument('trigger_id')
+    trigger_get_parser.add_argument('--cwd', default='.')
+    trigger_create_parser = subparsers.add_parser('trigger-create', help='create a local remote trigger')
+    trigger_create_parser.add_argument('--body-json', required=True)
+    trigger_create_parser.add_argument('--cwd', default='.')
+    trigger_update_parser = subparsers.add_parser('trigger-update', help='update a local remote trigger')
+    trigger_update_parser.add_argument('trigger_id')
+    trigger_update_parser.add_argument('--body-json', required=True)
+    trigger_update_parser.add_argument('--cwd', default='.')
+    trigger_run_parser = subparsers.add_parser('trigger-run', help='run a local remote trigger')
+    trigger_run_parser.add_argument('trigger_id')
+    trigger_run_parser.add_argument('--body-json', default='{}')
+    trigger_run_parser.add_argument('--cwd', default='.')
     teams_status_parser = subparsers.add_parser('team-status', help='show local collaboration team runtime summary')
     teams_status_parser.add_argument('--cwd', default='.')
     teams_list_parser = subparsers.add_parser('team-list', help='list local collaboration teams')
@@ -906,6 +946,33 @@ def main(argv: list[str] | None = None) -> int:
         runtime = RemoteRuntime.from_workspace(Path(args.cwd).resolve())
         print(runtime.disconnect().as_text())
         return 0
+    if args.command == 'worktree-status':
+        runtime = WorktreeRuntime.from_workspace(Path(args.cwd).resolve())
+        print('# Worktree')
+        print()
+        print(runtime.render_summary())
+        return 0
+    if args.command == 'worktree-enter':
+        runtime = WorktreeRuntime.from_workspace(Path(args.cwd).resolve())
+        try:
+            print(runtime.enter(name=args.name).as_text())
+        except (RuntimeError, ValueError) as exc:
+            print(exc)
+            return 1
+        return 0
+    if args.command == 'worktree-exit':
+        runtime = WorktreeRuntime.from_workspace(Path(args.cwd).resolve())
+        try:
+            print(
+                runtime.exit(
+                    action=args.action,
+                    discard_changes=args.discard_changes,
+                ).as_text()
+            )
+        except (RuntimeError, ValueError) as exc:
+            print(exc)
+            return 1
+        return 0
     if args.command == 'account-status':
         runtime = AccountRuntime.from_workspace(Path(args.cwd).resolve())
         print('# Account')
@@ -1038,6 +1105,80 @@ def main(argv: list[str] | None = None) -> int:
         print(f'effective_key_count={mutation.effective_key_count}')
         print(runtime.render_value(args.key_path))
         return 0
+    if args.command == 'workflow-list':
+        runtime = WorkflowRuntime.from_workspace(Path(args.cwd).resolve())
+        print(runtime.render_workflows_index(query=args.query))
+        return 0
+    if args.command == 'workflow-get':
+        runtime = WorkflowRuntime.from_workspace(Path(args.cwd).resolve())
+        try:
+            print(runtime.render_workflow(args.workflow_name))
+        except KeyError:
+            print(f'Unknown workflow: {args.workflow_name}')
+            return 1
+        return 0
+    if args.command == 'workflow-run':
+        runtime = WorkflowRuntime.from_workspace(Path(args.cwd).resolve())
+        arguments = json.loads(args.arguments_json)
+        if not isinstance(arguments, dict):
+            print('arguments-json must decode to a JSON object')
+            return 1
+        try:
+            print(runtime.render_run_report(args.workflow_name, arguments=arguments))
+        except KeyError:
+            print(f'Unknown workflow: {args.workflow_name}')
+            return 1
+        return 0
+    if args.command == 'trigger-list':
+        runtime = RemoteTriggerRuntime.from_workspace(Path(args.cwd).resolve())
+        print(runtime.render_trigger_index(query=args.query))
+        return 0
+    if args.command == 'trigger-get':
+        runtime = RemoteTriggerRuntime.from_workspace(Path(args.cwd).resolve())
+        try:
+            print(runtime.render_trigger(args.trigger_id))
+        except KeyError:
+            print(f'Unknown remote trigger: {args.trigger_id}')
+            return 1
+        return 0
+    if args.command == 'trigger-create':
+        runtime = RemoteTriggerRuntime.from_workspace(Path(args.cwd).resolve())
+        body = json.loads(args.body_json)
+        if not isinstance(body, dict):
+            print('body-json must decode to a JSON object')
+            return 1
+        try:
+            trigger = runtime.create_trigger(body)
+        except (KeyError, TypeError, ValueError) as exc:
+            print(exc)
+            return 1
+        print(runtime.render_trigger(trigger.trigger_id))
+        return 0
+    if args.command == 'trigger-update':
+        runtime = RemoteTriggerRuntime.from_workspace(Path(args.cwd).resolve())
+        body = json.loads(args.body_json)
+        if not isinstance(body, dict):
+            print('body-json must decode to a JSON object')
+            return 1
+        try:
+            trigger = runtime.update_trigger(args.trigger_id, body)
+        except (KeyError, TypeError, ValueError) as exc:
+            print(exc)
+            return 1
+        print(runtime.render_trigger(trigger.trigger_id))
+        return 0
+    if args.command == 'trigger-run':
+        runtime = RemoteTriggerRuntime.from_workspace(Path(args.cwd).resolve())
+        body = json.loads(args.body_json)
+        if not isinstance(body, dict):
+            print('body-json must decode to a JSON object')
+            return 1
+        try:
+            print(runtime.render_run_report(args.trigger_id, body=body))
+        except KeyError:
+            print(f'Unknown remote trigger: {args.trigger_id}')
+            return 1
+        return 0
     if args.command == 'team-status':
         runtime = TeamRuntime.from_workspace(Path(args.cwd).resolve())
         print('# Teams')

src/remote_trigger_runtime.py

@@ -0,0 +1,371 @@
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+
+DEFAULT_REMOTE_TRIGGER_STATE_PATH = Path('.port_sessions') / 'remote_trigger_runtime.json'
+REMOTE_TRIGGER_MANIFEST_FILES = ('.claw-remote-triggers.json', '.claw-triggers.json')
+
+
+@dataclass(frozen=True)
+class RemoteTriggerDefinition:
+    trigger_id: str
+    source: str
+    name: str | None = None
+    description: str | None = None
+    schedule: str | None = None
+    workflow: str | None = None
+    remote_target: str | None = None
+    body: dict[str, Any] = field(default_factory=dict)
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class RemoteTriggerRunRecord:
+    run_id: str
+    trigger_id: str
+    created_at: str
+    status: str
+    body: dict[str, Any] = field(default_factory=dict)
+    workflow: str | None = None
+    remote_target: str | None = None
+
+
+@dataclass
+class RemoteTriggerRuntime:
+    cwd: Path
+    triggers: tuple[RemoteTriggerDefinition, ...] = field(default_factory=tuple)
+    manifests: tuple[str, ...] = field(default_factory=tuple)
+    history: tuple[RemoteTriggerRunRecord, ...] = field(default_factory=tuple)
+    state_path: Path = field(default_factory=lambda: DEFAULT_REMOTE_TRIGGER_STATE_PATH.resolve())
+
+    @classmethod
+    def from_workspace(
+        cls,
+        cwd: Path,
+        additional_working_directories: tuple[str, ...] = (),
+    ) -> 'RemoteTriggerRuntime':
+        resolved_cwd = cwd.resolve()
+        manifest_paths = _discover_manifest_paths(resolved_cwd, additional_working_directories)
+        manifest_triggers: dict[str, RemoteTriggerDefinition] = {}
+        for manifest_path in manifest_paths:
+            for trigger in _load_triggers_from_manifest(manifest_path):
+                manifest_triggers[trigger.trigger_id] = trigger
+        state_path = resolved_cwd / DEFAULT_REMOTE_TRIGGER_STATE_PATH
+        payload = _load_payload(state_path)
+        custom_payload = payload.get('custom_triggers')
+        custom_triggers: dict[str, RemoteTriggerDefinition] = {}
+        if isinstance(custom_payload, list):
+            for item in custom_payload:
+                trigger = _definition_from_payload(item)
+                if trigger is not None:
+                    custom_triggers[trigger.trigger_id] = trigger
+        merged = {**manifest_triggers, **custom_triggers}
+        history_payload = payload.get('history')
+        history = tuple(
+            _run_record_from_payload(item)
+            for item in history_payload
+            if isinstance(item, dict) and _run_record_from_payload(item) is not None
+        ) if isinstance(history_payload, list) else ()
+        return cls(
+            cwd=resolved_cwd,
+            triggers=tuple(sorted(merged.values(), key=lambda item: item.trigger_id.lower())),
+            manifests=tuple(str(path) for path in manifest_paths),
+            history=history,
+            state_path=state_path,
+        )
+
+    def has_state(self) -> bool:
+        return bool(self.triggers or self.history)
+
+    def list_triggers(
+        self,
+        *,
+        query: str | None = None,
+        limit: int | None = None,
+    ) -> tuple[RemoteTriggerDefinition, ...]:
+        triggers = self.triggers
+        if query:
+            needle = query.lower()
+            triggers = tuple(
+                trigger
+                for trigger in triggers
+                if needle in trigger.trigger_id.lower()
+                or needle in (trigger.name or '').lower()
+                or needle in (trigger.description or '').lower()
+                or needle in (trigger.workflow or '').lower()
+            )
+        if limit is not None and limit >= 0:
+            triggers = triggers[:limit]
+        return triggers
+
+    def get_trigger(self, trigger_id: str) -> RemoteTriggerDefinition | None:
+        needle = trigger_id.strip().lower()
+        if not needle:
+            return None
+        for trigger in self.triggers:
+            if trigger.trigger_id.lower() == needle:
+                return trigger
+        return None
+
+    def create_trigger(self, body: dict[str, Any]) -> RemoteTriggerDefinition:
+        trigger = _definition_from_body(body, source='local_state')
+        if trigger.trigger_id in {item.trigger_id for item in self.triggers}:
+            raise KeyError(trigger.trigger_id)
+        self.triggers = tuple(sorted((*self.triggers, trigger), key=lambda item: item.trigger_id.lower()))
+        self._persist_state()
+        return trigger
+
+    def update_trigger(self, trigger_id: str, body: dict[str, Any]) -> RemoteTriggerDefinition:
+        existing = self.get_trigger(trigger_id)
+        if existing is None:
+            raise KeyError(trigger_id)
+        merged_body = {
+            'trigger_id': existing.trigger_id,
+            'name': existing.name,
+            'description': existing.description,
+            'schedule': existing.schedule,
+            'workflow': existing.workflow,
+            'remote_target': existing.remote_target,
+            'body': dict(existing.body),
+            'metadata': dict(existing.metadata),
+        }
+        merged_body.update(body)
+        if isinstance(body.get('body'), dict):
+            merged_body['body'] = dict(body['body'])
+        trigger = _definition_from_body(merged_body, source='local_state')
+        self.triggers = tuple(
+            sorted(
+                (trigger if item.trigger_id == trigger_id else item for item in self.triggers),
+                key=lambda item: item.trigger_id.lower(),
+            )
+        )
+        self._persist_state()
+        return trigger
+
+    def run_trigger(
+        self,
+        trigger_id: str,
+        *,
+        body: dict[str, Any] | None = None,
+    ) -> RemoteTriggerRunRecord:
+        trigger = self.get_trigger(trigger_id)
+        if trigger is None:
+            raise KeyError(trigger_id)
+        merged_body = dict(trigger.body)
+        if body:
+            merged_body.update(body)
+        record = RemoteTriggerRunRecord(
+            run_id=f'trigger_run_{uuid4().hex[:10]}',
+            trigger_id=trigger.trigger_id,
+            created_at=_utc_now(),
+            status='queued',
+            body=merged_body,
+            workflow=trigger.workflow,
+            remote_target=trigger.remote_target,
+        )
+        self.history = (*self.history, record)
+        self._persist_state()
+        return record
+
+    def render_summary(self) -> str:
+        lines = [
+            f'Local remote trigger manifests: {len(self.manifests)}',
+            f'Configured remote triggers: {len(self.triggers)}',
+            f'Remote trigger run history: {len(self.history)}',
+        ]
+        if self.triggers:
+            lines.append('- Latest triggers:')
+            for trigger in self.triggers[:5]:
+                label = trigger.name or trigger.trigger_id
+                workflow = trigger.workflow or 'none'
+                lines.append(f'  - {label}: workflow={workflow}')
+        return '\n'.join(lines)
+
+    def render_trigger_index(self, *, query: str | None = None) -> str:
+        triggers = self.list_triggers(query=query, limit=100)
+        lines = ['# Remote Triggers', '']
+        if not triggers:
+            lines.append('No local remote triggers discovered.')
+            return '\n'.join(lines)
+        for trigger in triggers:
+            label = trigger.name or trigger.trigger_id
+            details = [trigger.trigger_id, label]
+            if trigger.schedule:
+                details.append(f'schedule={trigger.schedule}')
+            if trigger.workflow:
+                details.append(f'workflow={trigger.workflow}')
+            if trigger.remote_target:
+                details.append(f'remote_target={trigger.remote_target}')
+            lines.append('- ' + ' ; '.join(details))
+        return '\n'.join(lines)
+
+    def render_trigger(self, trigger_id: str) -> str:
+        trigger = self.get_trigger(trigger_id)
+        if trigger is None:
+            raise KeyError(trigger_id)
+        lines = ['# Remote Trigger', '', f'trigger_id={trigger.trigger_id}']
+        lines.append(f'source={trigger.source}')
+        if trigger.name:
+            lines.append(f'name={trigger.name}')
+        if trigger.description:
+            lines.append(f'description={trigger.description}')
+        if trigger.schedule:
+            lines.append(f'schedule={trigger.schedule}')
+        if trigger.workflow:
+            lines.append(f'workflow={trigger.workflow}')
+        if trigger.remote_target:
+            lines.append(f'remote_target={trigger.remote_target}')
+        lines.append('body=' + json.dumps(trigger.body, ensure_ascii=True, sort_keys=True))
+        return '\n'.join(lines)
+
+    def render_run_report(
+        self,
+        trigger_id: str,
+        *,
+        body: dict[str, Any] | None = None,
+    ) -> str:
+        record = self.run_trigger(trigger_id, body=body)
+        lines = ['# Remote Trigger Run', '', f'run_id={record.run_id}', f'trigger_id={record.trigger_id}']
+        lines.append(f'status={record.status}')
+        lines.append(f'created_at={record.created_at}')
+        if record.workflow:
+            lines.append(f'workflow={record.workflow}')
+        if record.remote_target:
+            lines.append(f'remote_target={record.remote_target}')
+        lines.append('body=' + json.dumps(record.body, indent=2, ensure_ascii=True, sort_keys=True))
+        return '\n'.join(lines)
+
+    def _persist_state(self) -> None:
+        self.state_path.parent.mkdir(parents=True, exist_ok=True)
+        payload = {
+            'custom_triggers': [
+                asdict(trigger)
+                for trigger in self.triggers
+                if trigger.source == 'local_state'
+            ],
+            'history': [asdict(record) for record in self.history[-128:]],
+        }
+        self.state_path.write_text(
+            json.dumps(payload, indent=2, ensure_ascii=True),
+            encoding='utf-8',
+        )
+
+
+def _discover_manifest_paths(cwd: Path, additional_working_directories: tuple[str, ...]) -> tuple[Path, ...]:
+    directories = [cwd, *(Path(path).resolve() for path in additional_working_directories)]
+    seen: set[Path] = set()
+    found: list[Path] = []
+    for directory in directories:
+        for filename in REMOTE_TRIGGER_MANIFEST_FILES:
+            candidate = (directory / filename).resolve()
+            if candidate in seen or not candidate.exists():
+                continue
+            seen.add(candidate)
+            found.append(candidate)
+    return tuple(found)
+
+
+def _load_triggers_from_manifest(manifest_path: Path) -> list[RemoteTriggerDefinition]:
+    try:
+        payload = json.loads(manifest_path.read_text(encoding='utf-8'))
+    except (OSError, json.JSONDecodeError):
+        return []
+    raw_triggers = payload.get('triggers') if isinstance(payload, dict) else None
+    if not isinstance(raw_triggers, list):
+        return []
+    triggers: list[RemoteTriggerDefinition] = []
+    for item in raw_triggers:
+        if not isinstance(item, dict):
+            continue
+        trigger = _definition_from_body(item, source=str(manifest_path))
+        triggers.append(trigger)
+    return triggers
+
+
+def _load_payload(path: Path) -> dict[str, Any]:
+    if not path.exists():
+        return {}
+    try:
+        payload = json.loads(path.read_text(encoding='utf-8'))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    return payload if isinstance(payload, dict) else {}
+
+
+def _definition_from_payload(payload: dict[str, Any]) -> RemoteTriggerDefinition | None:
+    try:
+        return _definition_from_body(payload, source=str(payload.get('source') or 'local_state'))
+    except (TypeError, ValueError):
+        return None
+
+
+def _definition_from_body(body: dict[str, Any], *, source: str) -> RemoteTriggerDefinition:
+    trigger_id = body.get('trigger_id') or body.get('id') or body.get('name')
+    if not isinstance(trigger_id, str) or not trigger_id.strip():
+        raise ValueError('trigger body must define trigger_id, id, or name')
+    raw_body = body.get('body')
+    if raw_body is not None and not isinstance(raw_body, dict):
+        raise TypeError('body must be a JSON object when provided')
+    metadata = body.get('metadata')
+    if metadata is not None and not isinstance(metadata, dict):
+        raise TypeError('metadata must be a JSON object when provided')
+    return RemoteTriggerDefinition(
+        trigger_id=trigger_id.strip(),
+        source=source,
+        name=body['name'].strip() if isinstance(body.get('name'), str) and body['name'].strip() else None,
+        description=(
+            body['description'].strip()
+            if isinstance(body.get('description'), str) and body['description'].strip()
+            else None
+        ),
+        schedule=(
+            body['schedule'].strip()
+            if isinstance(body.get('schedule'), str) and body['schedule'].strip()
+            else None
+        ),
+        workflow=(
+            body['workflow'].strip()
+            if isinstance(body.get('workflow'), str) and body['workflow'].strip()
+            else None
+        ),
+        remote_target=(
+            body['remote_target'].strip()
+            if isinstance(body.get('remote_target'), str) and body['remote_target'].strip()
+            else None
+        ),
+        body=dict(raw_body or {}),
+        metadata=dict(metadata or {}),
+    )
+
+
+def _run_record_from_payload(payload: dict[str, Any]) -> RemoteTriggerRunRecord | None:
+    run_id = payload.get('run_id')
+    trigger_id = payload.get('trigger_id')
+    created_at = payload.get('created_at')
+    status = payload.get('status')
+    if not all(isinstance(value, str) and value for value in (run_id, trigger_id, created_at, status)):
+        return None
+    body = payload.get('body')
+    return RemoteTriggerRunRecord(
+        run_id=run_id,
+        trigger_id=trigger_id,
+        created_at=created_at,
+        status=status,
+        body=dict(body) if isinstance(body, dict) else {},
+        workflow=str(payload['workflow']) if isinstance(payload.get('workflow'), str) else None,
+        remote_target=(
+            str(payload['remote_target'])
+            if isinstance(payload.get('remote_target'), str)
+            else None
+        ),
+    )
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()

src/workflow_runtime.py

@@ -0,0 +1,319 @@
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+
+DEFAULT_WORKFLOW_STATE_PATH = Path('.port_sessions') / 'workflow_runtime.json'
+WORKFLOW_MANIFEST_FILES = ('.claw-workflows.json', '.claw-workflow.json')
+
+
+@dataclass(frozen=True)
+class WorkflowDefinition:
+    name: str
+    source_manifest: str
+    description: str | None = None
+    prompt: str | None = None
+    steps: tuple[dict[str, Any], ...] = ()
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class WorkflowRunRecord:
+    run_id: str
+    workflow_name: str
+    status: str
+    created_at: str
+    arguments: dict[str, Any] = field(default_factory=dict)
+    summary: str | None = None
+
+
+@dataclass
+class WorkflowRuntime:
+    cwd: Path
+    workflows: tuple[WorkflowDefinition, ...] = field(default_factory=tuple)
+    manifests: tuple[str, ...] = field(default_factory=tuple)
+    history: tuple[WorkflowRunRecord, ...] = field(default_factory=tuple)
+    state_path: Path = field(default_factory=lambda: DEFAULT_WORKFLOW_STATE_PATH.resolve())
+
+    @classmethod
+    def from_workspace(
+        cls,
+        cwd: Path,
+        additional_working_directories: tuple[str, ...] = (),
+    ) -> 'WorkflowRuntime':
+        resolved_cwd = cwd.resolve()
+        manifest_paths = _discover_manifest_paths(resolved_cwd, additional_working_directories)
+        workflows: list[WorkflowDefinition] = []
+        for manifest_path in manifest_paths:
+            workflows.extend(_load_workflows_from_manifest(manifest_path))
+        state_path = resolved_cwd / DEFAULT_WORKFLOW_STATE_PATH
+        payload = _load_payload(state_path)
+        history_payload = payload.get('history')
+        history = tuple(
+            _run_record_from_payload(item)
+            for item in history_payload
+            if isinstance(item, dict) and _run_record_from_payload(item) is not None
+        ) if isinstance(history_payload, list) else ()
+        return cls(
+            cwd=resolved_cwd,
+            workflows=tuple(workflows),
+            manifests=tuple(str(path) for path in manifest_paths),
+            history=history,
+            state_path=state_path,
+        )
+
+    def has_workflows(self) -> bool:
+        return bool(self.workflows or self.history)
+
+    def list_workflows(
+        self,
+        *,
+        query: str | None = None,
+        limit: int | None = None,
+    ) -> tuple[WorkflowDefinition, ...]:
+        workflows = self.workflows
+        if query:
+            needle = query.lower()
+            workflows = tuple(
+                workflow
+                for workflow in workflows
+                if needle in workflow.name.lower()
+                or needle in (workflow.description or '').lower()
+            )
+        if limit is not None and limit >= 0:
+            workflows = workflows[:limit]
+        return workflows
+
+    def get_workflow(self, name: str) -> WorkflowDefinition | None:
+        needle = name.strip().lower()
+        if not needle:
+            return None
+        for workflow in self.workflows:
+            if workflow.name.lower() == needle:
+                return workflow
+        return None
+
+    def run_workflow(
+        self,
+        name: str,
+        *,
+        arguments: dict[str, Any] | None = None,
+    ) -> WorkflowRunRecord:
+        workflow = self.get_workflow(name)
+        if workflow is None:
+            raise KeyError(name)
+        normalized_arguments = dict(arguments or {})
+        rendered_steps = _render_steps(workflow.steps, normalized_arguments)
+        summary = (
+            rendered_steps[0]
+            if rendered_steps
+            else workflow.description
+            or f'Workflow {workflow.name} recorded without steps.'
+        )
+        record = WorkflowRunRecord(
+            run_id=f'workflow_run_{uuid4().hex[:10]}',
+            workflow_name=workflow.name,
+            status='recorded',
+            created_at=_utc_now(),
+            arguments=normalized_arguments,
+            summary=summary,
+        )
+        self.history = (*self.history, record)
+        self._persist_state()
+        return record
+
+    def render_summary(self) -> str:
+        lines = [
+            f'Local workflow manifests: {len(self.manifests)}',
+            f'Configured workflows: {len(self.workflows)}',
+            f'Workflow run history: {len(self.history)}',
+        ]
+        if self.workflows:
+            lines.append('- Latest workflows:')
+            for workflow in self.workflows[:5]:
+                description = workflow.description or 'No description.'
+                lines.append(f'  - {workflow.name}: {description}')
+        return '\n'.join(lines)
+
+    def render_workflows_index(self, *, query: str | None = None) -> str:
+        workflows = self.list_workflows(query=query, limit=100)
+        lines = ['# Workflows', '']
+        if not workflows:
+            lines.append('No local workflows discovered.')
+            return '\n'.join(lines)
+        for workflow in workflows:
+            description = workflow.description or 'No description.'
+            lines.append(f'- {workflow.name} ; steps={len(workflow.steps)} ; {description}')
+        return '\n'.join(lines)
+
+    def render_workflow(self, name: str) -> str:
+        workflow = self.get_workflow(name)
+        if workflow is None:
+            raise KeyError(name)
+        lines = ['# Workflow', '', f'name={workflow.name}']
+        lines.append(f'source_manifest={workflow.source_manifest}')
+        lines.append(f'step_count={len(workflow.steps)}')
+        if workflow.description:
+            lines.extend(['', '## Description', workflow.description])
+        if workflow.prompt:
+            lines.extend(['', '## Prompt', workflow.prompt])
+        if workflow.steps:
+            lines.extend(['', '## Steps'])
+            for index, step in enumerate(workflow.steps, start=1):
+                title = step.get('title') or step.get('name') or f'Step {index}'
+                detail = step.get('detail') or step.get('command') or step.get('prompt') or ''
+                lines.append(f'{index}. {title}')
+                if detail:
+                    lines.append(f'   {detail}')
+        return '\n'.join(lines)
+
+    def render_run_report(
+        self,
+        name: str,
+        *,
+        arguments: dict[str, Any] | None = None,
+    ) -> str:
+        workflow = self.get_workflow(name)
+        if workflow is None:
+            raise KeyError(name)
+        normalized_arguments = dict(arguments or {})
+        record = self.run_workflow(name, arguments=normalized_arguments)
+        lines = ['# Workflow Run', '', f'run_id={record.run_id}', f'workflow={record.workflow_name}']
+        lines.append(f'status={record.status}')
+        lines.append(f'created_at={record.created_at}')
+        if normalized_arguments:
+            lines.extend(['', '## Arguments', json.dumps(normalized_arguments, indent=2, ensure_ascii=True)])
+        rendered_steps = _render_steps(workflow.steps, normalized_arguments)
+        if rendered_steps:
+            lines.extend(['', '## Resolved Steps'])
+            lines.extend(f'- {step}' for step in rendered_steps)
+        if workflow.prompt:
+            lines.extend(['', '## Prompt', _safe_format(workflow.prompt, normalized_arguments)])
+        return '\n'.join(lines)
+
+    def _persist_state(self) -> None:
+        self.state_path.parent.mkdir(parents=True, exist_ok=True)
+        payload = {
+            'history': [asdict(record) for record in self.history[-128:]],
+        }
+        self.state_path.write_text(
+            json.dumps(payload, indent=2, ensure_ascii=True),
+            encoding='utf-8',
+        )
+
+
+def _discover_manifest_paths(cwd: Path, additional_working_directories: tuple[str, ...]) -> tuple[Path, ...]:
+    directories = [cwd, *(Path(path).resolve() for path in additional_working_directories)]
+    seen: set[Path] = set()
+    found: list[Path] = []
+    for directory in directories:
+        for filename in WORKFLOW_MANIFEST_FILES:
+            candidate = (directory / filename).resolve()
+            if candidate in seen or not candidate.exists():
+                continue
+            seen.add(candidate)
+            found.append(candidate)
+    return tuple(found)
+
+
+def _load_workflows_from_manifest(manifest_path: Path) -> list[WorkflowDefinition]:
+    try:
+        payload = json.loads(manifest_path.read_text(encoding='utf-8'))
+    except (OSError, json.JSONDecodeError):
+        return []
+    items: list[dict[str, Any]] = []
+    if isinstance(payload, dict):
+        raw_workflows = payload.get('workflows')
+        if isinstance(raw_workflows, list):
+            items.extend(item for item in raw_workflows if isinstance(item, dict))
+        raw_workflow = payload.get('workflow')
+        if isinstance(raw_workflow, dict):
+            items.append(raw_workflow)
+    workflows: list[WorkflowDefinition] = []
+    for item in items:
+        name = item.get('name')
+        if not isinstance(name, str) or not name.strip():
+            continue
+        raw_steps = item.get('steps')
+        steps: list[dict[str, Any]] = []
+        if isinstance(raw_steps, list):
+            for entry in raw_steps:
+                if isinstance(entry, str):
+                    steps.append({'title': entry})
+                elif isinstance(entry, dict):
+                    steps.append(dict(entry))
+        workflows.append(
+            WorkflowDefinition(
+                name=name.strip(),
+                source_manifest=str(manifest_path),
+                description=(
+                    item['description'].strip()
+                    if isinstance(item.get('description'), str) and item['description'].strip()
+                    else None
+                ),
+                prompt=(
+                    item['prompt'].strip()
+                    if isinstance(item.get('prompt'), str) and item['prompt'].strip()
+                    else None
+                ),
+                steps=tuple(steps),
+                metadata=dict(item.get('metadata', {})) if isinstance(item.get('metadata'), dict) else {},
+            )
+        )
+    return workflows
+
+
+def _load_payload(path: Path) -> dict[str, Any]:
+    if not path.exists():
+        return {}
+    try:
+        payload = json.loads(path.read_text(encoding='utf-8'))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    return payload if isinstance(payload, dict) else {}
+
+
+def _run_record_from_payload(payload: dict[str, Any]) -> WorkflowRunRecord | None:
+    run_id = payload.get('run_id')
+    workflow_name = payload.get('workflow_name')
+    status = payload.get('status')
+    created_at = payload.get('created_at')
+    if not all(isinstance(value, str) and value for value in (run_id, workflow_name, status, created_at)):
+        return None
+    arguments = payload.get('arguments')
+    return WorkflowRunRecord(
+        run_id=run_id,
+        workflow_name=workflow_name,
+        status=status,
+        created_at=created_at,
+        arguments=dict(arguments) if isinstance(arguments, dict) else {},
+        summary=str(payload['summary']) if isinstance(payload.get('summary'), str) else None,
+    )
+
+
+def _render_steps(steps: tuple[dict[str, Any], ...], arguments: dict[str, Any]) -> list[str]:
+    rendered: list[str] = []
+    for index, step in enumerate(steps, start=1):
+        title = step.get('title') or step.get('name') or f'Step {index}'
+        detail = step.get('detail') or step.get('command') or step.get('prompt')
+        text = str(title)
+        if isinstance(detail, str) and detail.strip():
+            text = f'{text}: {_safe_format(detail, arguments)}'
+        rendered.append(_safe_format(text, arguments))
+    return rendered
+
+
+def _safe_format(template: str, arguments: dict[str, Any]) -> str:
+    text = template
+    for key, value in arguments.items():
+        text = text.replace('{' + str(key) + '}', str(value))
+    return text
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()

src/worktree_runtime.py

@@ -0,0 +1,448 @@
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+DEFAULT_WORKTREE_STATE_FILE = 'claw_worktree_runtime.json'
+DEFAULT_WORKTREE_PARENT_SUFFIX = '-claw-worktrees'
+VALID_EXIT_ACTIONS = ('keep', 'remove')
+_WORKTREE_SLUG_RE = re.compile(r'^[A-Za-z0-9._-]+(?:/[A-Za-z0-9._-]+)*$')
+
+
+@dataclass(frozen=True)
+class WorktreeSessionState:
+    name: str
+    repo_root: str
+    common_git_dir: str
+    worktree_path: str
+    worktree_branch: str
+    original_cwd: str
+    original_head_commit: str | None
+    created_at: str
+
+
+@dataclass(frozen=True)
+class WorktreeStatusReport:
+    active: bool
+    detail: str
+    repo_root: str | None = None
+    common_git_dir: str | None = None
+    current_cwd: str | None = None
+    original_cwd: str | None = None
+    worktree_path: str | None = None
+    worktree_branch: str | None = None
+    session_name: str | None = None
+    state_path: str | None = None
+    history_count: int = 0
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    def as_text(self) -> str:
+        lines = [
+            f'active={self.active}',
+            f'detail={self.detail}',
+            f'history_count={self.history_count}',
+        ]
+        for key in (
+            'repo_root',
+            'common_git_dir',
+            'current_cwd',
+            'original_cwd',
+            'worktree_path',
+            'worktree_branch',
+            'session_name',
+            'state_path',
+        ):
+            value = getattr(self, key)
+            if value:
+                lines.append(f'{key}={value}')
+        for key, value in sorted(self.metadata.items()):
+            lines.append(f'metadata.{key}={value}')
+        return '\n'.join(lines)
+
+
+@dataclass
+class WorktreeRuntime:
+    cwd: Path
+    state_path: Path
+    repo_root: Path | None = None
+    common_git_dir: Path | None = None
+    active_session: WorktreeSessionState | None = None
+    history: tuple[dict[str, Any], ...] = field(default_factory=tuple)
+
+    @classmethod
+    def from_workspace(cls, cwd: Path) -> 'WorktreeRuntime':
+        resolved_cwd = cwd.resolve()
+        common_git_dir = _find_git_common_dir(resolved_cwd)
+        repo_root = _infer_repo_root(resolved_cwd, common_git_dir)
+        state_path = (
+            common_git_dir / DEFAULT_WORKTREE_STATE_FILE
+            if common_git_dir is not None
+            else (resolved_cwd / '.port_sessions' / 'worktree_runtime.json')
+        )
+        payload = _load_payload(state_path)
+        active_payload = payload.get('active_session')
+        history_payload = payload.get('history')
+        return cls(
+            cwd=resolved_cwd,
+            state_path=state_path,
+            repo_root=repo_root,
+            common_git_dir=common_git_dir,
+            active_session=_session_from_payload(active_payload),
+            history=tuple(
+                item for item in history_payload if isinstance(item, dict)
+            ) if isinstance(history_payload, list) else (),
+        )
+
+    def has_state(self) -> bool:
+        return self.active_session is not None or bool(self.history)
+
+    def current_report(self, *, detail: str | None = None) -> WorktreeStatusReport:
+        if self.active_session is None:
+            return WorktreeStatusReport(
+                active=False,
+                detail=detail or 'No active managed worktree session.',
+                repo_root=str(self.repo_root) if self.repo_root is not None else None,
+                common_git_dir=(
+                    str(self.common_git_dir) if self.common_git_dir is not None else None
+                ),
+                current_cwd=str(self.cwd),
+                state_path=str(self.state_path),
+                history_count=len(self.history),
+            )
+        active = self.active_session
+        return WorktreeStatusReport(
+            active=True,
+            detail=detail or f'Active worktree session {active.name}',
+            repo_root=active.repo_root,
+            common_git_dir=active.common_git_dir,
+            current_cwd=str(self.cwd),
+            original_cwd=active.original_cwd,
+            worktree_path=active.worktree_path,
+            worktree_branch=active.worktree_branch,
+            session_name=active.name,
+            state_path=str(self.state_path),
+            history_count=len(self.history),
+        )
+
+    def render_summary(self) -> str:
+        lines = ['# Worktree', '']
+        report = self.current_report()
+        lines.extend(
+            [
+                f'- Git repo detected: {self.repo_root is not None}',
+                f'- Active managed worktree: {report.active}',
+                f'- Current working directory: {self.cwd}',
+                f'- Worktree history entries: {len(self.history)}',
+            ]
+        )
+        if self.repo_root is not None:
+            lines.append(f'- Repo root: {self.repo_root}')
+        if self.common_git_dir is not None:
+            lines.append(f'- Common git dir: {self.common_git_dir}')
+        if self.active_session is None:
+            lines.append('- Active worktree path: none')
+            lines.append('- Original working directory: none')
+        else:
+            active = self.active_session
+            lines.append(f'- Active worktree path: {active.worktree_path}')
+            lines.append(f'- Active worktree branch: {active.worktree_branch}')
+            lines.append(f'- Original working directory: {active.original_cwd}')
+        return '\n'.join(lines)
+
+    def render_history(self) -> str:
+        lines = ['# Worktree History', '']
+        if not self.history:
+            lines.append('No worktree history recorded.')
+            return '\n'.join(lines)
+        for entry in self.history:
+            action = entry.get('action', 'unknown')
+            timestamp = entry.get('timestamp', 'unknown')
+            name = entry.get('name') or entry.get('worktree_name') or 'unknown'
+            path = entry.get('worktree_path', 'unknown')
+            lines.append(f'- {timestamp} ; {action} ; {name} ; {path}')
+        return '\n'.join(lines)
+
+    def enter(self, name: str | None = None) -> WorktreeStatusReport:
+        if self.active_session is not None:
+            raise RuntimeError('A managed worktree session is already active.')
+        if self.repo_root is None or self.common_git_dir is None:
+            raise RuntimeError('A git repository is required to create a managed worktree.')
+        slug = _normalize_slug(name)
+        worktree_parent = self.repo_root.parent / f'{self.repo_root.name}{DEFAULT_WORKTREE_PARENT_SUFFIX}'
+        worktree_path = (worktree_parent / slug).resolve()
+        branch = f'claw/{slug}'
+        if worktree_path.exists():
+            raise RuntimeError(f'Worktree path already exists: {worktree_path}')
+        if _branch_exists(self.repo_root, branch):
+            raise RuntimeError(f'Worktree branch already exists: {branch}')
+        worktree_path.parent.mkdir(parents=True, exist_ok=True)
+        _run_git(
+            self.repo_root,
+            ['worktree', 'add', '-b', branch, str(worktree_path), 'HEAD'],
+        )
+        active = WorktreeSessionState(
+            name=slug,
+            repo_root=str(self.repo_root),
+            common_git_dir=str(self.common_git_dir),
+            worktree_path=str(worktree_path),
+            worktree_branch=branch,
+            original_cwd=str(self.cwd),
+            original_head_commit=_git_head(self.repo_root),
+            created_at=_utc_now(),
+        )
+        self.active_session = active
+        self.cwd = Path(active.worktree_path)
+        self._append_history(
+            {
+                'action': 'enter',
+                'timestamp': active.created_at,
+                'name': active.name,
+                'repo_root': active.repo_root,
+                'original_cwd': active.original_cwd,
+                'worktree_path': active.worktree_path,
+                'worktree_branch': active.worktree_branch,
+            }
+        )
+        self._persist_state()
+        return self.current_report(
+            detail=(
+                f'Created worktree at {active.worktree_path} on branch {active.worktree_branch}. '
+                'The session should now work inside the managed worktree.'
+            )
+        )
+
+    def exit(
+        self,
+        *,
+        action: str = 'keep',
+        discard_changes: bool = False,
+    ) -> WorktreeStatusReport:
+        normalized_action = action.strip().lower()
+        if normalized_action not in VALID_EXIT_ACTIONS:
+            raise ValueError(f'action must be one of {", ".join(VALID_EXIT_ACTIONS)}')
+        active = self.active_session
+        if active is None:
+            raise RuntimeError('No managed worktree session is currently active.')
+        worktree_path = Path(active.worktree_path)
+        if normalized_action == 'remove':
+            change_summary = _count_worktree_changes(
+                worktree_path,
+                active.original_head_commit,
+            )
+            if change_summary is None and not discard_changes:
+                raise RuntimeError(
+                    'Could not verify worktree cleanliness. Re-run with discard_changes=true to remove it.'
+                )
+            if change_summary is not None:
+                changed_files, commits = change_summary
+                if (changed_files > 0 or commits > 0) and not discard_changes:
+                    raise RuntimeError(
+                        'Worktree has uncommitted files or commits. '
+                        'Re-run with discard_changes=true to remove it.'
+                    )
+            _run_git(Path(active.repo_root), ['worktree', 'remove', '--force', active.worktree_path])
+            _run_git(Path(active.repo_root), ['branch', '-D', active.worktree_branch], check=False)
+        self._append_history(
+            {
+                'action': f'exit_{normalized_action}',
+                'timestamp': _utc_now(),
+                'name': active.name,
+                'repo_root': active.repo_root,
+                'original_cwd': active.original_cwd,
+                'worktree_path': active.worktree_path,
+                'worktree_branch': active.worktree_branch,
+                'discard_changes': discard_changes,
+            }
+        )
+        self.active_session = None
+        self.cwd = Path(active.original_cwd)
+        self._persist_state()
+        return WorktreeStatusReport(
+            active=False,
+            detail=(
+                f'Exited managed worktree {active.name} and returned to {active.original_cwd}.'
+                if normalized_action == 'keep'
+                else (
+                    f'Removed managed worktree {active.name} and returned to {active.original_cwd}.'
+                )
+            ),
+            repo_root=active.repo_root,
+            common_git_dir=active.common_git_dir,
+            current_cwd=active.original_cwd,
+            original_cwd=active.original_cwd,
+            worktree_path=active.worktree_path,
+            worktree_branch=active.worktree_branch,
+            session_name=active.name,
+            state_path=str(self.state_path),
+            history_count=len(self.history),
+            metadata={
+                'action': normalized_action,
+                'discard_changes': discard_changes,
+            },
+        )
+
+    def _append_history(self, entry: dict[str, Any]) -> None:
+        self.history = (*self.history, dict(entry))
+
+    def _persist_state(self) -> None:
+        self.state_path.parent.mkdir(parents=True, exist_ok=True)
+        payload = {
+            'active_session': (
+                asdict(self.active_session) if self.active_session is not None else None
+            ),
+            'history': [dict(entry) for entry in self.history[-64:]],
+        }
+        self.state_path.write_text(
+            json.dumps(payload, indent=2, ensure_ascii=True),
+            encoding='utf-8',
+        )
+
+
+def _normalize_slug(raw_name: str | None) -> str:
+    if raw_name is None or not raw_name.strip():
+        return f'worktree-{datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")}'
+    candidate = raw_name.strip()
+    if len(candidate) > 64 or not _WORKTREE_SLUG_RE.match(candidate):
+        raise ValueError(
+            'worktree name may contain only letters, digits, dots, underscores, dashes, '
+            'and optional "/" separators, with a maximum length of 64 characters'
+        )
+    return candidate
+
+
+def _find_git_common_dir(cwd: Path) -> Path | None:
+    try:
+        completed = subprocess.run(
+            ['git', '-C', str(cwd), 'rev-parse', '--git-common-dir'],
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+    except (OSError, subprocess.CalledProcessError):
+        return None
+    output = completed.stdout.strip()
+    if not output:
+        return None
+    candidate = Path(output)
+    if not candidate.is_absolute():
+        candidate = (cwd / candidate).resolve()
+    return candidate.resolve()
+
+
+def _infer_repo_root(cwd: Path, common_git_dir: Path | None) -> Path | None:
+    if common_git_dir is not None and common_git_dir.name == '.git':
+        return common_git_dir.parent.resolve()
+    try:
+        completed = subprocess.run(
+            ['git', '-C', str(cwd), 'rev-parse', '--show-toplevel'],
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+    except (OSError, subprocess.CalledProcessError):
+        return None
+    output = completed.stdout.strip()
+    return Path(output).resolve() if output else None
+
+
+def _load_payload(state_path: Path) -> dict[str, Any]:
+    if not state_path.exists():
+        return {}
+    try:
+        payload = json.loads(state_path.read_text(encoding='utf-8'))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    return payload if isinstance(payload, dict) else {}
+
+
+def _session_from_payload(payload: Any) -> WorktreeSessionState | None:
+    if not isinstance(payload, dict):
+        return None
+    required_keys = (
+        'name',
+        'repo_root',
+        'common_git_dir',
+        'worktree_path',
+        'worktree_branch',
+        'original_cwd',
+        'created_at',
+    )
+    if not all(isinstance(payload.get(key), str) and payload.get(key) for key in required_keys):
+        return None
+    return WorktreeSessionState(
+        name=str(payload['name']),
+        repo_root=str(payload['repo_root']),
+        common_git_dir=str(payload['common_git_dir']),
+        worktree_path=str(payload['worktree_path']),
+        worktree_branch=str(payload['worktree_branch']),
+        original_cwd=str(payload['original_cwd']),
+        original_head_commit=(
+            str(payload['original_head_commit'])
+            if isinstance(payload.get('original_head_commit'), str)
+            else None
+        ),
+        created_at=str(payload['created_at']),
+    )
+
+
+def _run_git(repo_root: Path, arguments: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ['git', '-C', str(repo_root), *arguments],
+        check=check,
+        capture_output=True,
+        text=True,
+    )
+
+
+def _branch_exists(repo_root: Path, branch: str) -> bool:
+    completed = _run_git(
+        repo_root,
+        ['show-ref', '--verify', '--quiet', f'refs/heads/{branch}'],
+        check=False,
+    )
+    return completed.returncode == 0
+
+
+def _git_head(repo_root: Path) -> str | None:
+    completed = _run_git(repo_root, ['rev-parse', 'HEAD'], check=False)
+    if completed.returncode != 0:
+        return None
+    value = completed.stdout.strip()
+    return value or None
+
+
+def _count_worktree_changes(worktree_path: Path, original_head_commit: str | None) -> tuple[int, int] | None:
+    status = subprocess.run(
+        ['git', '-C', str(worktree_path), 'status', '--porcelain'],
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if status.returncode != 0:
+        return None
+    changed_files = len([line for line in status.stdout.splitlines() if line.strip()])
+    if not original_head_commit:
+        return None
+    rev_list = subprocess.run(
+        ['git', '-C', str(worktree_path), 'rev-list', '--count', f'{original_head_commit}..HEAD'],
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if rev_list.returncode != 0:
+        return None
+    try:
+        commits = int(rev_list.stdout.strip() or '0')
+    except ValueError:
+        commits = 0
+    return changed_files, commits
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()

tests/test_agent_slash_commands.py

@@ -224,6 +224,41 @@ class AgentSlashCommandTests(unittest.TestCase):
         self.assertIn('profile=staging', ssh_result.final_output)
         self.assertIn('connected=False', disconnect_result.final_output)
 
+    def test_workflow_and_trigger_commands_render_local_reports(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            workspace = Path(tmp_dir)
+            (workspace / '.claw-workflows.json').write_text(
+                (
+                    '{"workflows":['
+                    '{"name":"review","description":"Review changes.","steps":["Inspect diff","Summarize findings"]}'
+                    ']}'
+                ),
+                encoding='utf-8',
+            )
+            (workspace / '.claw-triggers.json').write_text(
+                (
+                    '{"triggers":['
+                    '{"trigger_id":"nightly","name":"Nightly","workflow":"review","schedule":"0 0 * * *"}'
+                    ']}'
+                ),
+                encoding='utf-8',
+            )
+            agent = LocalCodingAgent(
+                model_config=ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct'),
+                runtime_config=AgentRuntimeConfig(cwd=workspace),
+            )
+            workflows_result = agent.run('/workflows')
+            workflow_result = agent.run('/workflow review')
+            trigger_result = agent.run('/trigger nightly')
+            trigger_run_result = agent.run('/trigger run nightly')
+        self.assertIn('# Workflows', workflows_result.final_output)
+        self.assertIn('review', workflows_result.final_output)
+        self.assertIn('# Workflow', workflow_result.final_output)
+        self.assertIn('Review changes', workflow_result.final_output)
+        self.assertIn('# Remote Trigger', trigger_result.final_output)
+        self.assertIn('trigger_id=nightly', trigger_result.final_output)
+        self.assertIn('# Remote Trigger Run', trigger_run_result.final_output)
+
     def test_account_commands_render_and_update_local_account_runtime(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_dir:
             workspace = Path(tmp_dir)

tests/test_main.py

@@ -154,6 +154,28 @@ class MainCliTests(unittest.TestCase):
         self.assertEqual(args.provider, 'local-search')
         self.assertEqual(args.cwd, '.')
 
+    def test_parser_accepts_worktree_runtime_commands(self) -> None:
+        parser = build_parser()
+        args = parser.parse_args(['worktree-exit', '--action', 'remove', '--discard-changes', '--cwd', '.'])
+        self.assertEqual(args.command, 'worktree-exit')
+        self.assertEqual(args.action, 'remove')
+        self.assertTrue(args.discard_changes)
+        self.assertEqual(args.cwd, '.')
+
+    def test_parser_accepts_workflow_runtime_commands(self) -> None:
+        parser = build_parser()
+        args = parser.parse_args(['workflow-run', 'review', '--arguments-json', '{"path":"src"}', '--cwd', '.'])
+        self.assertEqual(args.command, 'workflow-run')
+        self.assertEqual(args.workflow_name, 'review')
+        self.assertEqual(args.cwd, '.')
+
+    def test_parser_accepts_remote_trigger_runtime_commands(self) -> None:
+        parser = build_parser()
+        args = parser.parse_args(['trigger-run', 'nightly', '--body-json', '{"depth":"quick"}', '--cwd', '.'])
+        self.assertEqual(args.command, 'trigger-run')
+        self.assertEqual(args.trigger_id, 'nightly')
+        self.assertEqual(args.cwd, '.')
+
     def test_parser_accepts_mcp_runtime_commands(self) -> None:
         parser = build_parser()
         args = parser.parse_args(['mcp-tools', '--cwd', '.', '--server', 'remote'])

tests/test_remote_trigger_runtime.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from src.agent_tools import build_tool_context, default_tool_registry, execute_tool
+from src.agent_types import AgentPermissions, AgentRuntimeConfig
+from src.remote_trigger_runtime import RemoteTriggerRuntime
+
+
+class RemoteTriggerRuntimeTests(unittest.TestCase):
+    def test_remote_trigger_runtime_discovers_and_runs_trigger(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            workspace = Path(tmp_dir)
+            (workspace / '.claw-triggers.json').write_text(
+                (
+                    '{"triggers":['
+                    '{"trigger_id":"nightly","name":"Nightly","workflow":"review",'
+                    '"schedule":"0 0 * * *","body":{"depth":"full"}}'
+                    ']}'
+                ),
+                encoding='utf-8',
+            )
+            runtime = RemoteTriggerRuntime.from_workspace(workspace)
+            trigger_report = runtime.render_trigger('nightly')
+            run_report = runtime.render_run_report('nightly', body={'depth': 'quick'})
+
+        self.assertIn('trigger_id=nightly', trigger_report)
+        self.assertIn('workflow=review', run_report)
+        self.assertIn('"depth": "quick"', run_report)
+
+    def test_remote_trigger_tool_supports_create_update_run(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            workspace = Path(tmp_dir)
+            runtime = RemoteTriggerRuntime.from_workspace(workspace)
+            context = build_tool_context(
+                AgentRuntimeConfig(
+                    cwd=workspace,
+                    permissions=AgentPermissions(allow_file_write=True),
+                ),
+                remote_trigger_runtime=runtime,
+            )
+            create_result = execute_tool(
+                default_tool_registry(),
+                'remote_trigger',
+                {
+                    'action': 'create',
+                    'body': {
+                        'trigger_id': 'nightly',
+                        'name': 'Nightly',
+                        'workflow': 'review',
+                        'body': {'depth': 'full'},
+                    },
+                },
+                context,
+            )
+            update_result = execute_tool(
+                default_tool_registry(),
+                'remote_trigger',
+                {
+                    'action': 'update',
+                    'trigger_id': 'nightly',
+                    'body': {'schedule': '0 0 * * *'},
+                },
+                context,
+            )
+            run_result = execute_tool(
+                default_tool_registry(),
+                'remote_trigger',
+                {
+                    'action': 'run',
+                    'trigger_id': 'nightly',
+                    'body': {'depth': 'quick'},
+                },
+                context,
+            )
+
+        self.assertTrue(create_result.ok)
+        self.assertEqual(create_result.metadata.get('trigger_id'), 'nightly')
+        self.assertTrue(update_result.ok)
+        self.assertEqual(update_result.metadata.get('remote_trigger_action'), 'update')
+        self.assertTrue(run_result.ok)
+        self.assertIn('# Remote Trigger Run', run_result.content)

tests/test_workflow_runtime.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from src.agent_tools import build_tool_context, default_tool_registry, execute_tool
+from src.agent_types import AgentRuntimeConfig
+from src.workflow_runtime import WorkflowRuntime
+
+
+class WorkflowRuntimeTests(unittest.TestCase):
+    def test_workflow_runtime_discovers_and_runs_workflow(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            workspace = Path(tmp_dir)
+            (workspace / '.claw-workflows.json').write_text(
+                (
+                    '{"workflows":['
+                    '{"name":"review","description":"Review the current patch.",'
+                    '"steps":[{"title":"Inspect diff","detail":"Read {path}"},{"title":"Summarize"}],'
+                    '"prompt":"Review changes under {path}"}'
+                    ']}'
+                ),
+                encoding='utf-8',
+            )
+            runtime = WorkflowRuntime.from_workspace(workspace)
+            rendered = runtime.render_workflow('review')
+            run_report = runtime.render_run_report('review', arguments={'path': 'src/'})
+
+        self.assertIn('Review the current patch', rendered)
+        self.assertIn('Read src/', run_report)
+        self.assertIn('Review changes under src/', run_report)
+
+    def test_workflow_tools_execute_against_runtime(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            workspace = Path(tmp_dir)
+            (workspace / '.claw-workflows.json').write_text(
+                (
+                    '{"workflows":['
+                    '{"name":"build","description":"Build the project.",'
+                    '"steps":["Inspect package","Run build"]}'
+                    ']}'
+                ),
+                encoding='utf-8',
+            )
+            runtime = WorkflowRuntime.from_workspace(workspace)
+            context = build_tool_context(
+                AgentRuntimeConfig(cwd=workspace),
+                workflow_runtime=runtime,
+            )
+            list_result = execute_tool(
+                default_tool_registry(),
+                'workflow_list',
+                {},
+                context,
+            )
+            get_result = execute_tool(
+                default_tool_registry(),
+                'workflow_get',
+                {'workflow_name': 'build'},
+                context,
+            )
+            run_result = execute_tool(
+                default_tool_registry(),
+                'workflow_run',
+                {'workflow_name': 'build', 'arguments': {'target': 'dist'}},
+                context,
+            )
+
+        self.assertTrue(list_result.ok)
+        self.assertIn('build', list_result.content)
+        self.assertTrue(get_result.ok)
+        self.assertIn('Build the project', get_result.content)
+        self.assertTrue(run_result.ok)
+        self.assertEqual(run_result.metadata.get('action'), 'workflow_run')
+        self.assertIn('# Workflow Run', run_result.content)
+

tests/test_worktree_runtime.py

@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from src.agent_runtime import LocalCodingAgent
+from src.agent_tools import build_tool_context, default_tool_registry, execute_tool
+from src.agent_types import AgentPermissions, AgentRuntimeConfig, ModelConfig
+from src.worktree_runtime import WorktreeRuntime
+
+
+class _FakeHTTPResponse:
+    def __init__(self, payload: dict[str, object]) -> None:
+        self.payload = payload
+
+    def read(self) -> bytes:
+        return json.dumps(self.payload).encode('utf-8')
+
+    def __enter__(self) -> '_FakeHTTPResponse':
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        return None
+
+
+def _make_urlopen_side_effect(responses: list[dict[str, object]]):
+    queued = [_FakeHTTPResponse(payload) for payload in responses]
+
+    def _fake_urlopen(request_obj, timeout=None):  # noqa: ANN001
+        return queued.pop(0)
+
+    return _fake_urlopen
+
+
+def _init_git_repo(workspace: Path) -> None:
+    subprocess.run(['git', 'init', '-q'], cwd=workspace, check=True)
+    subprocess.run(['git', 'config', 'user.email', 'test@example.com'], cwd=workspace, check=True)
+    subprocess.run(['git', 'config', 'user.name', 'Test User'], cwd=workspace, check=True)
+    (workspace / 'README.md').write_text('hello\n', encoding='utf-8')
+    subprocess.run(['git', 'add', 'README.md'], cwd=workspace, check=True)
+    subprocess.run(['git', 'commit', '-qm', 'init'], cwd=workspace, check=True)
+
+
+@unittest.skipUnless(shutil.which('git'), 'git is required for worktree tests')
+class WorktreeRuntimeTests(unittest.TestCase):
+    def test_worktree_runtime_enters_and_exits_managed_session(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            workspace = Path(tmp_dir)
+            _init_git_repo(workspace)
+            runtime = WorktreeRuntime.from_workspace(workspace)
+            enter_report = runtime.enter('feature-preview')
+            worktree_path = Path(enter_report.worktree_path or '')
+            exit_report = runtime.exit(action='keep')
+
+        self.assertTrue(enter_report.active)
+        self.assertTrue(worktree_path.exists())
+        self.assertIn('feature-preview', enter_report.worktree_branch or '')
+        self.assertFalse(exit_report.active)
+        self.assertEqual(exit_report.original_cwd, str(workspace))
+
+    def test_worktree_tools_execute_against_runtime(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            workspace = Path(tmp_dir)
+            _init_git_repo(workspace)
+            runtime = WorktreeRuntime.from_workspace(workspace)
+            context = build_tool_context(
+                AgentRuntimeConfig(
+                    cwd=workspace,
+                    permissions=AgentPermissions(allow_file_write=True),
+                ),
+                worktree_runtime=runtime,
+            )
+            enter_result = execute_tool(
+                default_tool_registry(),
+                'worktree_enter',
+                {'name': 'preview'},
+                context,
+            )
+            status_result = execute_tool(
+                default_tool_registry(),
+                'worktree_status',
+                {},
+                context,
+            )
+            exit_result = execute_tool(
+                default_tool_registry(),
+                'worktree_exit',
+                {'action': 'remove', 'discard_changes': True},
+                context,
+            )
+
+        self.assertTrue(enter_result.ok)
+        self.assertIn('preview', enter_result.content)
+        self.assertEqual(enter_result.metadata.get('action'), 'worktree_enter')
+        self.assertTrue(status_result.ok)
+        self.assertIn('Active managed worktree: True', status_result.content)
+        self.assertTrue(exit_result.ok)
+        self.assertEqual(exit_result.metadata.get('action'), 'worktree_exit')
+
+    def test_agent_switches_cwd_after_worktree_enter(self) -> None:
+        responses = [
+            {
+                'choices': [
+                    {
+                        'message': {
+                            'role': 'assistant',
+                            'content': 'Entering worktree.',
+                            'tool_calls': [
+                                {
+                                    'id': 'call_enter',
+                                    'type': 'function',
+                                    'function': {
+                                        'name': 'worktree_enter',
+                                        'arguments': '{"name":"preview"}',
+                                    },
+                                }
+                            ],
+                        },
+                        'finish_reason': 'tool_calls',
+                    }
+                ],
+                'usage': {'prompt_tokens': 8, 'completion_tokens': 2},
+            },
+            {
+                'choices': [
+                    {
+                        'message': {
+                            'role': 'assistant',
+                            'content': 'Writing inside the worktree.',
+                            'tool_calls': [
+                                {
+                                    'id': 'call_write',
+                                    'type': 'function',
+                                    'function': {
+                                        'name': 'write_file',
+                                        'arguments': '{"path":"note.txt","content":"from worktree\\n"}',
+                                    },
+                                }
+                            ],
+                        },
+                        'finish_reason': 'tool_calls',
+                    }
+                ],
+                'usage': {'prompt_tokens': 8, 'completion_tokens': 2},
+            },
+            {
+                'choices': [
+                    {
+                        'message': {
+                            'role': 'assistant',
+                            'content': 'done',
+                        },
+                        'finish_reason': 'stop',
+                    }
+                ],
+                'usage': {'prompt_tokens': 6, 'completion_tokens': 1},
+            },
+        ]
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            workspace = Path(tmp_dir)
+            _init_git_repo(workspace)
+            with patch(
+                'src.openai_compat.request.urlopen',
+                side_effect=_make_urlopen_side_effect(responses),
+            ):
+                agent = LocalCodingAgent(
+                    model_config=ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct'),
+                    runtime_config=AgentRuntimeConfig(
+                        cwd=workspace,
+                        permissions=AgentPermissions(allow_file_write=True),
+                    ),
+                )
+                result = agent.run('Use a worktree and write a file there')
+                runtime = WorktreeRuntime.from_workspace(workspace)
+                assert runtime.active_session is not None
+                worktree_path = Path(runtime.active_session.worktree_path)
+
+        self.assertEqual(result.final_output, 'done')
+        self.assertFalse((workspace / 'note.txt').exists())
+        self.assertTrue((worktree_path / 'note.txt').exists())
+        self.assertEqual(agent.runtime_config.cwd, worktree_path.resolve())
+