add claw-code-agent/src/plan_runtime.py and claw-code-agent/src/background_runtime.py. The agent now has a real
persistent plan runtime with update_plan, plan_get, and plan_clear, plus plan-to-task sync wired through claw-code-agent/src/ agent_tools.py, claw-code-agent/src/agent_runtime.py, claw-code-agent/src/agent_context.py, claw-code-agent/src/agent_prompting.py, and claw-code-agent/src/ agent_slash_commands.py. New local plan slash commands are /plan and /planner.
This commit is contained in:
Abdelrahman Abdallah
@@ -5,13 +5,16 @@ import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from src.agent_context import (
|
||||
build_context_snapshot,
|
||||
clear_context_caches,
|
||||
set_system_prompt_injection,
|
||||
)
|
||||
from src.plan_runtime import PlanRuntime
|
||||
from src.agent_types import AgentRuntimeConfig
|
||||
from src.task_runtime import TaskRuntime
|
||||
|
||||
|
||||
class AgentContextTests(unittest.TestCase):
|
||||
@@ -57,6 +60,76 @@ class AgentContextTests(unittest.TestCase):
|
||||
self.assertIn('demo-plugin', snapshot.user_context['pluginCache'])
|
||||
self.assertIn('1.2.3', snapshot.user_context['pluginCache'])
|
||||
|
||||
def test_user_context_loads_hook_policy_manifest(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir) / 'repo'
|
||||
workspace.mkdir(parents=True)
|
||||
(workspace / '.claw-policy.json').write_text(
|
||||
(
|
||||
'{"trusted": false, '
|
||||
'"managedSettings": {"reviewMode": "strict"}, '
|
||||
'"safeEnv": ["HOOK_SAFE_TOKEN"], '
|
||||
'"hooks": {"beforePrompt": ["Respect workspace policy."]}}'
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
with patch.dict('os.environ', {'HOOK_SAFE_TOKEN': 'demo-secret'}, clear=False):
|
||||
snapshot = build_context_snapshot(AgentRuntimeConfig(cwd=workspace))
|
||||
|
||||
self.assertIn('hookPolicy', snapshot.user_context)
|
||||
self.assertIn('managedSettings', snapshot.user_context)
|
||||
self.assertIn('safeEnv', snapshot.user_context)
|
||||
self.assertIn('trustMode', snapshot.user_context)
|
||||
self.assertIn('reviewMode=strict', snapshot.user_context['managedSettings'])
|
||||
self.assertIn('HOOK_SAFE_TOKEN=demo-secret', snapshot.user_context['safeEnv'])
|
||||
self.assertIn('untrusted', snapshot.user_context['trustMode'])
|
||||
|
||||
def test_user_context_loads_mcp_runtime_summary(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir) / 'repo'
|
||||
workspace.mkdir(parents=True)
|
||||
(workspace / 'notes.txt').write_text('mcp notes\n', encoding='utf-8')
|
||||
(workspace / '.claw-mcp.json').write_text(
|
||||
(
|
||||
'{"servers":[{"name":"workspace","resources":['
|
||||
'{"uri":"mcp://workspace/notes","name":"Notes","path":"notes.txt"}'
|
||||
']}]}'
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
|
||||
snapshot = build_context_snapshot(AgentRuntimeConfig(cwd=workspace))
|
||||
|
||||
self.assertIn('mcpRuntime', snapshot.user_context)
|
||||
self.assertIn('Local MCP resources: 1', snapshot.user_context['mcpRuntime'])
|
||||
|
||||
def test_user_context_loads_task_runtime_summary(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir) / 'repo'
|
||||
workspace.mkdir(parents=True)
|
||||
runtime = TaskRuntime.from_workspace(workspace)
|
||||
runtime.create_task(title='Review task runtime')
|
||||
|
||||
snapshot = build_context_snapshot(AgentRuntimeConfig(cwd=workspace))
|
||||
|
||||
self.assertIn('taskRuntime', snapshot.user_context)
|
||||
self.assertIn('Total tasks: 1', snapshot.user_context['taskRuntime'])
|
||||
|
||||
def test_user_context_loads_plan_runtime_summary(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir) / 'repo'
|
||||
workspace.mkdir(parents=True)
|
||||
plan_runtime = PlanRuntime.from_workspace(workspace)
|
||||
plan_runtime.update_plan(
|
||||
[{'step': 'Inspect the runtime', 'status': 'in_progress'}],
|
||||
explanation='Use a stored plan.',
|
||||
)
|
||||
|
||||
snapshot = build_context_snapshot(AgentRuntimeConfig(cwd=workspace))
|
||||
|
||||
self.assertIn('planRuntime', snapshot.user_context)
|
||||
self.assertIn('Total plan steps: 1', snapshot.user_context['planRuntime'])
|
||||
|
||||
@unittest.skipIf(shutil.which('git') is None, 'git is required for git context tests')
|
||||
def test_git_status_snapshot_contains_branch_and_status(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
|
||||
@@ -5,10 +5,12 @@ import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from src.agent_prompting import build_prompt_context, build_system_prompt_parts, render_system_prompt
|
||||
from src.plan_runtime import PlanRuntime
|
||||
from src.agent_runtime import LocalCodingAgent
|
||||
from src.agent_session import AgentSessionState
|
||||
from src.agent_tools import default_tool_registry
|
||||
from src.agent_types import AgentPermissions, AgentRuntimeConfig, ModelConfig
|
||||
from src.task_runtime import TaskRuntime
|
||||
|
||||
|
||||
class AgentPromptingTests(unittest.TestCase):
|
||||
@@ -79,3 +81,83 @@ class AgentPromptingTests(unittest.TestCase):
|
||||
|
||||
prompt = render_system_prompt(parts)
|
||||
self.assertIn('# Plugins', prompt)
|
||||
|
||||
def test_prompt_builder_mentions_hook_policy_when_manifest_is_loaded(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
(workspace / '.claw-policy.json').write_text(
|
||||
'{"trusted": false, "hooks": {"beforePrompt": ["Follow workspace policy."]}}',
|
||||
encoding='utf-8',
|
||||
)
|
||||
runtime_config = AgentRuntimeConfig(cwd=workspace)
|
||||
model_config = ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct')
|
||||
prompt_context = build_prompt_context(runtime_config, model_config)
|
||||
parts = build_system_prompt_parts(
|
||||
prompt_context=prompt_context,
|
||||
runtime_config=runtime_config,
|
||||
tools=default_tool_registry(),
|
||||
)
|
||||
|
||||
prompt = render_system_prompt(parts)
|
||||
self.assertIn('# Hook Policy', prompt)
|
||||
|
||||
def test_prompt_builder_mentions_mcp_when_manifest_is_loaded(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
(workspace / 'notes.txt').write_text('mcp notes\n', encoding='utf-8')
|
||||
(workspace / '.claw-mcp.json').write_text(
|
||||
(
|
||||
'{"servers":[{"name":"workspace","resources":['
|
||||
'{"uri":"mcp://workspace/notes","name":"Notes","path":"notes.txt"}'
|
||||
']}]}'
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
runtime_config = AgentRuntimeConfig(cwd=workspace)
|
||||
model_config = ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct')
|
||||
prompt_context = build_prompt_context(runtime_config, model_config)
|
||||
parts = build_system_prompt_parts(
|
||||
prompt_context=prompt_context,
|
||||
runtime_config=runtime_config,
|
||||
tools=default_tool_registry(),
|
||||
)
|
||||
|
||||
prompt = render_system_prompt(parts)
|
||||
self.assertIn('# MCP', prompt)
|
||||
|
||||
def test_prompt_builder_mentions_tasks_when_runtime_is_loaded(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
runtime = TaskRuntime.from_workspace(workspace)
|
||||
runtime.create_task(title='Inspect runtime tasks')
|
||||
runtime_config = AgentRuntimeConfig(cwd=workspace)
|
||||
model_config = ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct')
|
||||
prompt_context = build_prompt_context(runtime_config, model_config)
|
||||
parts = build_system_prompt_parts(
|
||||
prompt_context=prompt_context,
|
||||
runtime_config=runtime_config,
|
||||
tools=default_tool_registry(),
|
||||
)
|
||||
|
||||
prompt = render_system_prompt(parts)
|
||||
self.assertIn('# Tasks', prompt)
|
||||
|
||||
def test_prompt_builder_mentions_planning_when_runtime_is_loaded(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
runtime = PlanRuntime.from_workspace(workspace)
|
||||
runtime.update_plan(
|
||||
[{'step': 'Inspect runtime planning', 'status': 'pending'}],
|
||||
explanation='Track the current plan.',
|
||||
)
|
||||
runtime_config = AgentRuntimeConfig(cwd=workspace)
|
||||
model_config = ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct')
|
||||
prompt_context = build_prompt_context(runtime_config, model_config)
|
||||
parts = build_system_prompt_parts(
|
||||
prompt_context=prompt_context,
|
||||
runtime_config=runtime_config,
|
||||
tools=default_tool_registry(),
|
||||
)
|
||||
|
||||
prompt = render_system_prompt(parts)
|
||||
self.assertIn('# Planning', prompt)
|
||||
|
||||
@@ -245,6 +245,26 @@ class AgentRuntimeTests(unittest.TestCase):
|
||||
)
|
||||
self.assertFalse(result.ok)
|
||||
self.assertIn('--allow-write', result.content)
|
||||
self.assertEqual(result.metadata.get('error_kind'), 'permission_denied')
|
||||
|
||||
def test_build_tool_context_supports_safe_env_overlay_for_shell_tools(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
config = AgentRuntimeConfig(
|
||||
cwd=Path(tmp_dir),
|
||||
permissions=AgentPermissions(allow_shell_commands=True),
|
||||
)
|
||||
context = build_tool_context(
|
||||
config,
|
||||
extra_env={'HOOK_SAFE_TOKEN': 'demo-secret'},
|
||||
)
|
||||
result = execute_tool(
|
||||
default_tool_registry(),
|
||||
'bash',
|
||||
{'command': 'printf %s "$HOOK_SAFE_TOKEN"'},
|
||||
context,
|
||||
)
|
||||
self.assertTrue(result.ok)
|
||||
self.assertIn('demo-secret', result.content)
|
||||
|
||||
def test_local_slash_command_returns_without_model_call(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
@@ -2379,3 +2399,151 @@ class AgentRuntimeTests(unittest.TestCase):
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
def test_hook_policy_before_prompt_injection_and_after_turn_event(self) -> None:
|
||||
responses = [
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'Completed under policy guidance.',
|
||||
},
|
||||
'finish_reason': 'stop',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 6, 'completion_tokens': 3},
|
||||
}
|
||||
]
|
||||
recorded_payloads: list[dict[str, object]] = []
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
(workspace / '.claw-policy.json').write_text(
|
||||
(
|
||||
'{"trusted": false, '
|
||||
'"managedSettings": {"reviewMode": "strict"}, '
|
||||
'"safeEnv": ["HOOK_SAFE_TOKEN"], '
|
||||
'"hooks": {"beforePrompt": ["Respect workspace policy."], '
|
||||
'"afterTurn": ["Persist the policy decision."]}}'
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
with patch.dict('os.environ', {'HOOK_SAFE_TOKEN': 'demo-secret'}, clear=False):
|
||||
with patch(
|
||||
'src.openai_compat.request.urlopen',
|
||||
side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads),
|
||||
):
|
||||
agent = LocalCodingAgent(
|
||||
model_config=ModelConfig(
|
||||
model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
|
||||
base_url='http://127.0.0.1:8000/v1',
|
||||
),
|
||||
runtime_config=AgentRuntimeConfig(cwd=workspace),
|
||||
)
|
||||
result = agent.run('Inspect the repository.')
|
||||
payload_text = json.dumps(recorded_payloads[0])
|
||||
self.assertIn('Workspace hook/policy guidance', payload_text)
|
||||
self.assertIn('Respect workspace policy.', payload_text)
|
||||
self.assertIn('Trust mode: untrusted', payload_text)
|
||||
self.assertEqual(agent.tool_context.extra_env.get('HOOK_SAFE_TOKEN'), 'demo-secret')
|
||||
self.assertIn('reviewMode=strict', agent.render_trust_report())
|
||||
hook_events = [event for event in result.events if event.get('type') == 'hook_policy_after_turn']
|
||||
self.assertEqual(len(hook_events), 1)
|
||||
self.assertEqual(hook_events[0].get('message'), 'Persist the policy decision.')
|
||||
|
||||
def test_hook_policy_blocks_tool_and_tracks_permission_denial(self) -> None:
|
||||
responses = [
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'Trying bash first.',
|
||||
'tool_calls': [
|
||||
{
|
||||
'id': 'call_1',
|
||||
'type': 'function',
|
||||
'function': {
|
||||
'name': 'bash',
|
||||
'arguments': '{"command": "pwd"}',
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
'finish_reason': 'tool_calls',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
|
||||
},
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'Bash was blocked by workspace policy.',
|
||||
},
|
||||
'finish_reason': 'stop',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 7, 'completion_tokens': 3},
|
||||
},
|
||||
]
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
(workspace / '.claw-policy.json').write_text(
|
||||
'{"denyTools": ["bash"]}',
|
||||
encoding='utf-8',
|
||||
)
|
||||
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',
|
||||
base_url='http://127.0.0.1:8000/v1',
|
||||
),
|
||||
runtime_config=AgentRuntimeConfig(cwd=workspace),
|
||||
)
|
||||
result = agent.run('Run pwd')
|
||||
self.assertEqual(result.final_output, 'Bash was blocked by workspace policy.')
|
||||
event_types = [event.get('type') for event in result.events]
|
||||
self.assertIn('hook_policy_tool_block', event_types)
|
||||
self.assertIn('tool_permission_denial', event_types)
|
||||
tool_message = next(
|
||||
message
|
||||
for message in result.transcript
|
||||
if message.get('role') == 'tool'
|
||||
)
|
||||
self.assertTrue(tool_message['metadata'].get('hook_policy_blocked'))
|
||||
self.assertEqual(tool_message['metadata'].get('error_kind'), 'permission_denied')
|
||||
|
||||
def test_hook_policy_budget_override_applies_when_runtime_budget_is_unset(self) -> None:
|
||||
responses = [
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'One model call happened.',
|
||||
},
|
||||
'finish_reason': 'stop',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 4, 'completion_tokens': 2},
|
||||
}
|
||||
]
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
(workspace / '.claw-policy.json').write_text(
|
||||
'{"budget": {"max_model_calls": 0}}',
|
||||
encoding='utf-8',
|
||||
)
|
||||
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',
|
||||
base_url='http://127.0.0.1:8000/v1',
|
||||
),
|
||||
runtime_config=AgentRuntimeConfig(cwd=workspace),
|
||||
)
|
||||
result = agent.run('Hello')
|
||||
self.assertEqual(result.stop_reason, 'budget_exceeded')
|
||||
self.assertIn('model-call budget was exceeded', result.final_output)
|
||||
|
||||
@@ -3,10 +3,13 @@ from __future__ import annotations
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from src.agent_runtime import LocalCodingAgent
|
||||
from src.agent_slash_commands import looks_like_command, parse_slash_command
|
||||
from src.agent_types import AgentRuntimeConfig, ModelConfig
|
||||
from src.plan_runtime import PlanRuntime
|
||||
from src.task_runtime import TaskRuntime
|
||||
|
||||
|
||||
class AgentSlashCommandTests(unittest.TestCase):
|
||||
@@ -53,6 +56,72 @@ class AgentSlashCommandTests(unittest.TestCase):
|
||||
self.assertIn('### Estimated usage by category', result.final_output)
|
||||
self.assertIn('### Memory Files', result.final_output)
|
||||
|
||||
def test_mcp_and_resource_commands_render_local_reports(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
(workspace / 'notes.txt').write_text('mcp notes\n', encoding='utf-8')
|
||||
(workspace / '.claw-mcp.json').write_text(
|
||||
(
|
||||
'{"servers":[{"name":"workspace","resources":['
|
||||
'{"uri":"mcp://workspace/notes","name":"Notes","path":"notes.txt"}'
|
||||
']}]}'
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
agent = LocalCodingAgent(
|
||||
model_config=ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct'),
|
||||
runtime_config=AgentRuntimeConfig(cwd=workspace),
|
||||
)
|
||||
mcp_result = agent.run('/mcp')
|
||||
resources_result = agent.run('/resources')
|
||||
resource_result = agent.run('/resource mcp://workspace/notes')
|
||||
legacy_mcp_result = agent.run('/mcp (MCP)')
|
||||
self.assertIn('# MCP', mcp_result.final_output)
|
||||
self.assertIn('Local MCP resources: 1', mcp_result.final_output)
|
||||
self.assertIn('# MCP Resources', resources_result.final_output)
|
||||
self.assertIn('mcp://workspace/notes', resources_result.final_output)
|
||||
self.assertIn('# MCP Resource', resource_result.final_output)
|
||||
self.assertIn('mcp notes', resource_result.final_output)
|
||||
self.assertIn('# MCP', legacy_mcp_result.final_output)
|
||||
|
||||
def test_tasks_and_task_commands_render_local_reports(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
runtime = TaskRuntime.from_workspace(workspace)
|
||||
mutation = runtime.create_task(
|
||||
title='Review runtime tasks',
|
||||
status='in_progress',
|
||||
)
|
||||
task_id = mutation.task.task_id if mutation.task is not None else ''
|
||||
agent = LocalCodingAgent(
|
||||
model_config=ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct'),
|
||||
runtime_config=AgentRuntimeConfig(cwd=workspace),
|
||||
)
|
||||
tasks_result = agent.run('/tasks')
|
||||
task_result = agent.run(f'/task {task_id}')
|
||||
todo_result = agent.run('/todo in_progress')
|
||||
self.assertIn('# Tasks', tasks_result.final_output)
|
||||
self.assertIn(task_id, tasks_result.final_output)
|
||||
self.assertIn('# Task', task_result.final_output)
|
||||
self.assertIn('in_progress', task_result.final_output)
|
||||
self.assertIn('# Tasks', todo_result.final_output)
|
||||
|
||||
def test_plan_command_renders_local_report(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
plan_runtime = PlanRuntime.from_workspace(workspace)
|
||||
plan_runtime.update_plan(
|
||||
[{'step': 'Inspect the plan command', 'status': 'in_progress'}],
|
||||
explanation='Use the local plan runtime.',
|
||||
)
|
||||
agent = LocalCodingAgent(
|
||||
model_config=ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct'),
|
||||
runtime_config=AgentRuntimeConfig(cwd=workspace),
|
||||
)
|
||||
plan_result = agent.run('/plan')
|
||||
self.assertIn('# Plan', plan_result.final_output)
|
||||
self.assertIn('Inspect the plan command', plan_result.final_output)
|
||||
|
||||
def test_tools_and_status_commands_render_local_reports(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
agent = LocalCodingAgent(
|
||||
@@ -66,6 +135,31 @@ class AgentSlashCommandTests(unittest.TestCase):
|
||||
self.assertIn('# Status', status_result.final_output)
|
||||
self.assertIn('Last run: none', status_result.final_output)
|
||||
|
||||
def test_hooks_and_trust_commands_render_local_reports(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
(workspace / '.claw-policy.json').write_text(
|
||||
(
|
||||
'{"trusted": false, '
|
||||
'"managedSettings": {"reviewMode": "strict"}, '
|
||||
'"safeEnv": ["HOOK_SAFE_TOKEN"]}'
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
with patch.dict('os.environ', {'HOOK_SAFE_TOKEN': 'demo-secret'}, clear=False):
|
||||
agent = LocalCodingAgent(
|
||||
model_config=ModelConfig(model='Qwen/Qwen3-Coder-30B-A3B-Instruct'),
|
||||
runtime_config=AgentRuntimeConfig(cwd=workspace),
|
||||
)
|
||||
hooks_result = agent.run('/hooks')
|
||||
trust_result = agent.run('/trust')
|
||||
self.assertIn('# Hook Policy', hooks_result.final_output)
|
||||
self.assertIn('Local hook/policy manifests', hooks_result.final_output)
|
||||
self.assertIn('# Trust', trust_result.final_output)
|
||||
self.assertIn('untrusted', trust_result.final_output)
|
||||
self.assertIn('reviewMode=strict', trust_result.final_output)
|
||||
self.assertIn('HOOK_SAFE_TOKEN=demo-secret', trust_result.final_output)
|
||||
|
||||
def test_clear_command_clears_saved_runtime_state(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
agent = LocalCodingAgent(
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from src.background_runtime import BackgroundSessionRuntime
|
||||
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
class BackgroundRuntimeTests(unittest.TestCase):
|
||||
def test_runtime_can_launch_and_kill_generic_process(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
runtime = BackgroundSessionRuntime(workspace / '.port_sessions' / 'background')
|
||||
record = runtime.launch(
|
||||
[sys.executable, '-c', 'import time; time.sleep(10)'],
|
||||
prompt='sleep',
|
||||
workspace_cwd=workspace,
|
||||
model='local/test-model',
|
||||
process_cwd=workspace,
|
||||
)
|
||||
running = runtime.load_record(record.background_id)
|
||||
killed = runtime.kill(record.background_id)
|
||||
for _ in range(30):
|
||||
if runtime.load_record(record.background_id).status != 'running':
|
||||
break
|
||||
time.sleep(0.1)
|
||||
|
||||
self.assertEqual(running.status, 'running')
|
||||
self.assertEqual(killed.status, 'killed')
|
||||
self.assertEqual(killed.stop_reason, 'killed')
|
||||
|
||||
def test_agent_background_cli_exposes_ps_logs_and_attach(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
run_dir = Path(tmp_dir)
|
||||
workspace = run_dir / 'workspace'
|
||||
workspace.mkdir()
|
||||
env = os.environ.copy()
|
||||
existing_pythonpath = env.get('PYTHONPATH')
|
||||
env['PYTHONPATH'] = (
|
||||
f'{PROJECT_ROOT}:{existing_pythonpath}'
|
||||
if existing_pythonpath
|
||||
else str(PROJECT_ROOT)
|
||||
)
|
||||
launch = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
'-m',
|
||||
'src.main',
|
||||
'agent-bg',
|
||||
'/help',
|
||||
'--cwd',
|
||||
str(workspace),
|
||||
],
|
||||
cwd=run_dir,
|
||||
env=env,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
background_id = next(
|
||||
line.split('=', 1)[1]
|
||||
for line in launch.stdout.splitlines()
|
||||
if line.startswith('background_id=')
|
||||
)
|
||||
runtime = BackgroundSessionRuntime(run_dir / '.port_sessions' / 'background')
|
||||
record = runtime.load_record(background_id)
|
||||
for _ in range(60):
|
||||
if record.status in {'completed', 'failed', 'exited'}:
|
||||
break
|
||||
time.sleep(0.1)
|
||||
record = runtime.load_record(background_id)
|
||||
ps = subprocess.run(
|
||||
[sys.executable, '-m', 'src.main', 'agent-ps'],
|
||||
cwd=run_dir,
|
||||
env=env,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
logs = subprocess.run(
|
||||
[sys.executable, '-m', 'src.main', 'agent-logs', background_id],
|
||||
cwd=run_dir,
|
||||
env=env,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
attach = subprocess.run(
|
||||
[sys.executable, '-m', 'src.main', 'agent-attach', background_id],
|
||||
cwd=run_dir,
|
||||
env=env,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
self.assertIn(background_id, launch.stdout)
|
||||
self.assertIn(background_id, ps.stdout)
|
||||
self.assertIn('# Background Logs', logs.stdout)
|
||||
self.assertIn('# Slash Commands', logs.stdout)
|
||||
self.assertIn('# Background Attach', attach.stdout)
|
||||
@@ -0,0 +1,158 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
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 AgentRuntimeConfig, ModelConfig
|
||||
from src.mcp_runtime import MCPRuntime
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
class MCPRuntimeTests(unittest.TestCase):
|
||||
def test_runtime_discovers_and_reads_local_resources(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
(workspace / 'notes.txt').write_text('mcp notes\n', encoding='utf-8')
|
||||
(workspace / '.claw-mcp.json').write_text(
|
||||
(
|
||||
'{"servers":[{"name":"workspace","resources":['
|
||||
'{"uri":"mcp://workspace/notes","name":"Notes","path":"notes.txt"},'
|
||||
'{"uri":"mcp://workspace/inline","name":"Inline","text":"inline body"}'
|
||||
']}]}'
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
runtime = MCPRuntime.from_workspace(workspace)
|
||||
self.assertEqual(len(runtime.resources), 2)
|
||||
self.assertIn('Local MCP resources: 2', runtime.render_summary())
|
||||
self.assertEqual(runtime.read_resource('mcp://workspace/inline'), 'inline body')
|
||||
self.assertIn('mcp notes', runtime.read_resource('mcp://workspace/notes'))
|
||||
|
||||
def test_mcp_tools_execute_against_runtime(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
(workspace / 'notes.txt').write_text('mcp notes\n', encoding='utf-8')
|
||||
(workspace / '.claw-mcp.json').write_text(
|
||||
(
|
||||
'{"servers":[{"name":"workspace","resources":['
|
||||
'{"uri":"mcp://workspace/notes","name":"Notes","path":"notes.txt"}'
|
||||
']}]}'
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
runtime = MCPRuntime.from_workspace(workspace)
|
||||
context = build_tool_context(
|
||||
AgentRuntimeConfig(cwd=workspace),
|
||||
mcp_runtime=runtime,
|
||||
)
|
||||
list_result = execute_tool(
|
||||
default_tool_registry(),
|
||||
'mcp_list_resources',
|
||||
{},
|
||||
context,
|
||||
)
|
||||
read_result = execute_tool(
|
||||
default_tool_registry(),
|
||||
'mcp_read_resource',
|
||||
{'uri': 'mcp://workspace/notes'},
|
||||
context,
|
||||
)
|
||||
|
||||
self.assertTrue(list_result.ok)
|
||||
self.assertIn('mcp://workspace/notes', list_result.content)
|
||||
self.assertTrue(read_result.ok)
|
||||
self.assertIn('mcp notes', read_result.content)
|
||||
|
||||
def test_agent_can_use_mcp_tools_in_model_loop(self) -> None:
|
||||
responses = [
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'I will inspect the MCP resource.',
|
||||
'tool_calls': [
|
||||
{
|
||||
'id': 'call_1',
|
||||
'type': 'function',
|
||||
'function': {
|
||||
'name': 'mcp_read_resource',
|
||||
'arguments': '{"uri": "mcp://workspace/notes"}',
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
'finish_reason': 'tool_calls',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
|
||||
},
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'The MCP resource says mcp notes.',
|
||||
},
|
||||
'finish_reason': 'stop',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 6, 'completion_tokens': 3},
|
||||
},
|
||||
]
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
(workspace / 'notes.txt').write_text('mcp notes\n', encoding='utf-8')
|
||||
(workspace / '.claw-mcp.json').write_text(
|
||||
(
|
||||
'{"servers":[{"name":"workspace","resources":['
|
||||
'{"uri":"mcp://workspace/notes","name":"Notes","path":"notes.txt"}'
|
||||
']}]}'
|
||||
),
|
||||
encoding='utf-8',
|
||||
)
|
||||
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',
|
||||
base_url='http://127.0.0.1:8000/v1',
|
||||
),
|
||||
runtime_config=AgentRuntimeConfig(cwd=workspace),
|
||||
)
|
||||
result = agent.run('Read the MCP notes resource')
|
||||
|
||||
self.assertEqual(result.final_output, 'The MCP resource says mcp notes.')
|
||||
self.assertEqual(result.tool_calls, 1)
|
||||
tool_message = next(
|
||||
message
|
||||
for message in result.transcript
|
||||
if message.get('role') == 'tool'
|
||||
)
|
||||
self.assertIn('mcp notes', tool_message.get('content', ''))
|
||||
@@ -0,0 +1,189 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
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.plan_runtime import PlanRuntime
|
||||
from src.task_runtime import TaskRuntime
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
class PlanRuntimeTests(unittest.TestCase):
|
||||
def test_runtime_persists_and_syncs_plan_to_tasks(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
task_runtime = TaskRuntime.from_workspace(workspace)
|
||||
plan_runtime = PlanRuntime.from_workspace(workspace)
|
||||
mutation = plan_runtime.update_plan(
|
||||
[
|
||||
{
|
||||
'step': 'Inspect the runtime loop',
|
||||
'status': 'in_progress',
|
||||
'description': 'Read the core agent files first.',
|
||||
},
|
||||
{
|
||||
'step': 'Patch the tool registry',
|
||||
'status': 'pending',
|
||||
},
|
||||
],
|
||||
explanation='Work through the runtime in two phases.',
|
||||
task_runtime=task_runtime,
|
||||
)
|
||||
rendered_plan = plan_runtime.render_plan()
|
||||
rendered_tasks = task_runtime.render_tasks()
|
||||
|
||||
self.assertEqual(mutation.after_count, 2)
|
||||
self.assertEqual(mutation.synced_tasks, 2)
|
||||
self.assertIn('Inspect the runtime loop', rendered_plan)
|
||||
self.assertIn('Work through the runtime in two phases.', rendered_plan)
|
||||
self.assertIn('Inspect the runtime loop', rendered_tasks)
|
||||
|
||||
def test_plan_tools_execute_against_runtime(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
task_runtime = TaskRuntime.from_workspace(workspace)
|
||||
plan_runtime = PlanRuntime.from_workspace(workspace)
|
||||
context = build_tool_context(
|
||||
AgentRuntimeConfig(
|
||||
cwd=workspace,
|
||||
permissions=AgentPermissions(allow_file_write=True),
|
||||
),
|
||||
plan_runtime=plan_runtime,
|
||||
task_runtime=task_runtime,
|
||||
)
|
||||
update_result = execute_tool(
|
||||
default_tool_registry(),
|
||||
'update_plan',
|
||||
{
|
||||
'explanation': 'Follow the current plan.',
|
||||
'items': [
|
||||
{'step': 'Inspect the workspace', 'status': 'completed'},
|
||||
{'step': 'Implement the fix', 'status': 'in_progress'},
|
||||
],
|
||||
},
|
||||
context,
|
||||
)
|
||||
get_result = execute_tool(
|
||||
default_tool_registry(),
|
||||
'plan_get',
|
||||
{},
|
||||
context,
|
||||
)
|
||||
clear_result = execute_tool(
|
||||
default_tool_registry(),
|
||||
'plan_clear',
|
||||
{'sync_tasks': True},
|
||||
context,
|
||||
)
|
||||
|
||||
self.assertTrue(update_result.ok)
|
||||
self.assertEqual(update_result.metadata.get('total_steps'), 2)
|
||||
self.assertEqual(update_result.metadata.get('synced_tasks'), 2)
|
||||
self.assertIn('# Plan', get_result.content)
|
||||
self.assertTrue(clear_result.ok)
|
||||
self.assertEqual(clear_result.metadata.get('total_steps'), 0)
|
||||
|
||||
def test_agent_can_use_update_plan_tool_in_model_loop(self) -> None:
|
||||
responses = [
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'I will store the plan first.',
|
||||
'tool_calls': [
|
||||
{
|
||||
'id': 'call_1',
|
||||
'type': 'function',
|
||||
'function': {
|
||||
'name': 'update_plan',
|
||||
'arguments': json.dumps(
|
||||
{
|
||||
'explanation': 'Start with a plan.',
|
||||
'items': [
|
||||
{
|
||||
'step': 'Inspect the current files',
|
||||
'status': 'in_progress',
|
||||
},
|
||||
{
|
||||
'step': 'Apply the code changes',
|
||||
'status': 'pending',
|
||||
},
|
||||
],
|
||||
}
|
||||
),
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
'finish_reason': 'tool_calls',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
|
||||
},
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'The plan was stored successfully.',
|
||||
},
|
||||
'finish_reason': 'stop',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 6, 'completion_tokens': 3},
|
||||
},
|
||||
]
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
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',
|
||||
base_url='http://127.0.0.1:8000/v1',
|
||||
),
|
||||
runtime_config=AgentRuntimeConfig(
|
||||
cwd=workspace,
|
||||
permissions=AgentPermissions(allow_file_write=True),
|
||||
),
|
||||
)
|
||||
result = agent.run('Store the current plan')
|
||||
self.assertTrue((workspace / '.port_sessions' / 'plan_runtime.json').exists())
|
||||
|
||||
self.assertEqual(result.final_output, 'The plan was stored successfully.')
|
||||
self.assertEqual(result.tool_calls, 1)
|
||||
tool_message = next(
|
||||
message for message in result.transcript if message.get('role') == 'tool'
|
||||
)
|
||||
self.assertIn('update_plan', tool_message.get('content', ''))
|
||||
@@ -0,0 +1,167 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
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.task_runtime import TaskRuntime
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
class TaskRuntimeTests(unittest.TestCase):
|
||||
def test_runtime_persists_and_renders_tasks(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
runtime = TaskRuntime.from_workspace(workspace)
|
||||
created = runtime.create_task(
|
||||
title='Implement task runtime',
|
||||
description='Add persistent tasks.',
|
||||
status='in_progress',
|
||||
)
|
||||
assert created.task is not None
|
||||
runtime.update_task(created.task.task_id, status='done')
|
||||
rendered_tasks = runtime.render_tasks()
|
||||
rendered_task = runtime.render_task(created.task.task_id)
|
||||
|
||||
self.assertIn('Implement task runtime', rendered_tasks)
|
||||
self.assertIn('done', rendered_task)
|
||||
|
||||
def test_task_tools_execute_against_runtime(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
runtime = TaskRuntime.from_workspace(workspace)
|
||||
context = build_tool_context(
|
||||
AgentRuntimeConfig(
|
||||
cwd=workspace,
|
||||
permissions=AgentPermissions(allow_file_write=True),
|
||||
),
|
||||
task_runtime=runtime,
|
||||
)
|
||||
create_result = execute_tool(
|
||||
default_tool_registry(),
|
||||
'task_create',
|
||||
{'title': 'Review task tools', 'status': 'todo'},
|
||||
context,
|
||||
)
|
||||
self.assertTrue(create_result.ok)
|
||||
task_id = str(create_result.metadata.get('task_id'))
|
||||
list_result = execute_tool(
|
||||
default_tool_registry(),
|
||||
'task_list',
|
||||
{},
|
||||
context,
|
||||
)
|
||||
get_result = execute_tool(
|
||||
default_tool_registry(),
|
||||
'task_get',
|
||||
{'task_id': task_id},
|
||||
context,
|
||||
)
|
||||
update_result = execute_tool(
|
||||
default_tool_registry(),
|
||||
'task_update',
|
||||
{'task_id': task_id, 'status': 'done'},
|
||||
context,
|
||||
)
|
||||
todo_result = execute_tool(
|
||||
default_tool_registry(),
|
||||
'todo_write',
|
||||
{'items': [{'title': 'Replace with todo snapshot', 'status': 'in_progress'}]},
|
||||
context,
|
||||
)
|
||||
|
||||
self.assertIn(task_id, list_result.content)
|
||||
self.assertIn('Review task tools', get_result.content)
|
||||
self.assertTrue(update_result.ok)
|
||||
self.assertEqual(update_result.metadata.get('task_status'), 'done')
|
||||
self.assertTrue(todo_result.ok)
|
||||
self.assertEqual(todo_result.metadata.get('total_tasks'), 1)
|
||||
|
||||
def test_agent_can_use_task_tools_in_model_loop(self) -> None:
|
||||
responses = [
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'I will create a task first.',
|
||||
'tool_calls': [
|
||||
{
|
||||
'id': 'call_1',
|
||||
'type': 'function',
|
||||
'function': {
|
||||
'name': 'task_create',
|
||||
'arguments': '{"title": "Review runtime tasks", "status": "todo"}',
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
'finish_reason': 'tool_calls',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 8, 'completion_tokens': 3},
|
||||
},
|
||||
{
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'The task was created successfully.',
|
||||
},
|
||||
'finish_reason': 'stop',
|
||||
}
|
||||
],
|
||||
'usage': {'prompt_tokens': 6, 'completion_tokens': 3},
|
||||
},
|
||||
]
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
workspace = Path(tmp_dir)
|
||||
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',
|
||||
base_url='http://127.0.0.1:8000/v1',
|
||||
),
|
||||
runtime_config=AgentRuntimeConfig(
|
||||
cwd=workspace,
|
||||
permissions=AgentPermissions(allow_file_write=True),
|
||||
),
|
||||
)
|
||||
result = agent.run('Create a task for the current work')
|
||||
self.assertTrue((workspace / '.port_sessions' / 'task_runtime.json').exists())
|
||||
|
||||
self.assertEqual(result.final_output, 'The task was created successfully.')
|
||||
self.assertEqual(result.tool_calls, 1)
|
||||
tool_message = next(
|
||||
message
|
||||
for message in result.transcript
|
||||
if message.get('role') == 'tool'
|
||||
)
|
||||
self.assertIn('task_create', tool_message.get('content', ''))
|
||||
Reference in New Issue
Block a user