Merge pull request #22 from rusiaaman/modes

Modes

gpt_action_json_schema.json

@@ -408,6 +408,47 @@
         ],
         "title": "BashInteractionWithUUID"
       },
+      "CodeWriterMode": {
+        "properties": {
+          "allowed_globs": {
+            "anyOf": [
+              {
+                "type": "string",
+                "const": "all"
+              },
+              {
+                "items": {
+                  "type": "string"
+                },
+                "type": "array"
+              }
+            ],
+            "title": "Allowed Globs"
+          },
+          "allowed_commands": {
+            "anyOf": [
+              {
+                "type": "string",
+                "const": "all"
+              },
+              {
+                "items": {
+                  "type": "string"
+                },
+                "type": "array"
+              }
+            ],
+            "title": "Allowed Commands"
+          }
+        },
+        "additionalProperties": false,
+        "type": "object",
+        "required": [
+          "allowed_globs",
+          "allowed_commands"
+        ],
+        "title": "CodeWriterMode"
+      },
       "CommandWithUUID": {
         "properties": {
           "command": {
@@ -520,6 +561,25 @@
             "type": "string",
             "title": "Task Id To Resume"
           },
+          "mode_name": {
+            "type": "string",
+            "enum": [
+              "wcgw",
+              "architect",
+              "code_writer"
+            ],
+            "title": "Mode Name"
+          },
+          "code_writer_config": {
+            "anyOf": [
+              {
+                "$ref": "#/components/schemas/CodeWriterMode"
+              },
+              {
+                "type": "null"
+              }
+            ]
+          },
           "user_id": {
             "type": "string",
             "format": "uuid",
@@ -532,6 +592,7 @@
           "any_workspace_path",
           "initial_files_to_read",
           "task_id_to_resume",
+          "mode_name",
           "user_id"
         ],
         "title": "InitializeWithUUID"

gpt_instructions.txt

@@ -15,6 +15,10 @@ Instructions for `Initialize`:
     - If the user has mentioned a folder or file with unclear project root, use the file or folder as `any_workspace_path`.
     - If user has mentioned any files use `initial_files_to_read` to read, use absolute paths only.
     - If `any_workspace_path` is provided, a tree structure of the workspace will be shown.
+    - Leave `any_workspace_path` as empty if no file or folder is mentioned.
+    - By default use mode `wcgw`
+    - In code-writer mode, set the commands and globs which user asked to set, otherwise use 'all'.
+    - In order to change the mode later, call this tool again but be sure to not provide any other argument like task_id_to_resume unnecessarily.
 
 Instructions for `BashCommand`:
 - Execute a bash command. This is stateful (beware with subsequent calls).

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 authors = [{ name = "Aman Rusia", email = "gapypi@arcfu.com" }]
 name = "wcgw"
-version = "2.7.2"
+version = "2.8.0"
 description = "Shell and coding agent on claude and chatgpt"
 readme = "README.md"
 requires-python = ">=3.11, <3.13"

src/wcgw/client/anthropic_client.py

@@ -130,7 +130,7 @@ def loop(
     memory = None
     if resume:
         try:
-            _, memory = load_memory(
+            _, memory, _ = load_memory(
                 resume,
                 8000,
                 lambda x: default_enc.encode(x).ids,
@@ -166,6 +166,7 @@ def loop(
 - The control will return to you in 5 seconds regardless of the status. For heavy commands, keep checking status using BashInteraction till they are finished.
 - Run long running commands in background using screen instead of "&".
 - Use longer wait_for_seconds if the command is expected to run for a long time.
+- Do not use 'cat' to read files, use ReadFiles tool instead.
 """,
         ),
         ToolParam(
@@ -281,22 +282,13 @@ def loop(
             ),
         ]
 
-    initial_info = initialize(
-        os.getcwd(), [], resume if (memory and resume) else "", 8000
+    system = initialize(
+        os.getcwd(),
+        [],
+        resume if (memory and resume) else "",
+        max_tokens=8000,
+        mode="wcgw",
     )
-    system = f"""
-You're an expert software engineer with shell and code knowledge.
-
-Instructions:
-
-    - You should use the provided bash execution, reading and writing file tools to complete objective.
-    - First understand about the project by getting the folder structure (ignoring .git, node_modules, venv, etc.)
-    - Always read relevant files before editing.
-    - Do not provide code snippets unless asked by the user, instead directly add/edit the code.
-    - Do not install new tools/packages before ensuring no such tools/package or an alternative already exists.
-
-{initial_info}
-"""
 
     with open(os.path.join(os.path.dirname(__file__), "diff-instructions.txt")) as f:
         system += f.read()

src/wcgw/client/mcp_server/server.py

@@ -28,6 +28,7 @@
 )
 from .. import tools
 from ..computer_use import SLEEP_TIME_MAX_S
+from ..modes import get_kt_prompt
 from ..tools import DoneFlag, default_enc, get_tool_output, which_tool_name
 
 COMPUTER_USE_ON_DOCKER_ENABLED = False
@@ -45,48 +46,35 @@ async def handle_read_resource(uri: AnyUrl) -> str:
     raise ValueError("No resources available")
 
 
-@server.list_prompts()  # type: ignore
-async def handle_list_prompts() -> list[types.Prompt]:
-    return [
+PROMPTS = {
+    "KnowledgeTransfer": (
         types.Prompt(
             name="KnowledgeTransfer",
             description="Prompt for invoking ContextSave tool in order to do a comprehensive knowledge transfer of a coding task. Prompts to save detailed error log and instructions.",
-        )
-    ]
+        ),
+        get_kt_prompt,
+    )
+}
+
+
+@server.list_prompts()  # type: ignore
+async def handle_list_prompts() -> list[types.Prompt]:
+    return [x[0] for x in PROMPTS.values()]
 
 
 @server.get_prompt()  # type: ignore
 async def handle_get_prompt(
     name: str, arguments: dict[str, str] | None
 ) -> types.GetPromptResult:
-    messages = []
-    if name == "KnowledgeTransfer":
-        messages = [
-            types.PromptMessage(
-                role="user",
-                content=types.TextContent(
-                    type="text",
-                    text="""Use `ContextSave` tool to do a knowledge transfer of the task in hand.
-Write detailed description in order to do a KT.
-Save all information necessary for a person to understand the task and the problems.
-
-Format the `description` field using Markdown with the following sections.
-- "# Objective" section containing project and task objective.
-- "# All user instructions" section should be provided containing all instructions user shared in the conversation.
-- "# Current status of the task" should be provided containing only what is already achieved, not what's remaining.
-- "# All issues with snippets" section containing snippets of error, traceback, file snippets, commands, etc. But no comments or solutions.
-- Be very verbose in the all issues with snippets section providing as much error context as possible.
-- "# Build and development instructions" section containing instructions to build or run project or run tests, or envrionment related information. Only include what's known. Leave empty if unknown.
-- After the tool completes succesfully, tell me the task id and the file path the tool generated (important!)
-- This tool marks end of your conversation, do not run any further tools after calling this.
-
-Provide all relevant file paths in order to understand and solve the the task. Err towards providing more file paths than fewer.
-
-(Note to self: this conversation can then be resumed later asking "Resume `<generated id>`" which should call Initialize tool)
-""",
-                ),
-            )
-        ]
+    messages = [
+        types.PromptMessage(
+            role="user",
+            content=types.TextContent(
+                type="text",
+                text=PROMPTS[name][1](),
+            ),
+        )
+    ]
     return types.GetPromptResult(messages=messages)
 
 
@@ -117,6 +105,9 @@ async def handle_list_tools() -> list[types.Tool]:
 - If user has mentioned any files use `initial_files_to_read` to read, use absolute paths only.
 - If `any_workspace_path` is provided, a tree structure of the workspace will be shown.
 - Leave `any_workspace_path` as empty if no file or folder is mentioned.
+- By default use mode "wcgw"
+- In "code-writer" mode, set the commands and globs which user asked to set, otherwise use 'all'.
+- In order to change the mode later, call this tool again but be sure to not provide any other argument like task_id_to_resume unnecessarily.
 """,
         ),
         ToolParam(
@@ -263,6 +254,8 @@ async def handle_call_tool(
     except ValidationError:
 
         def try_json(x: str) -> Any:
+            if not isinstance(x, str):
+                return x
             try:
                 return json.loads(x)
             except json.JSONDecodeError:
@@ -284,24 +277,7 @@ def try_json(x: str) -> Any:
         if isinstance(output_or_done, str):
             if issubclass(tool_type, Initialize):
                 output_or_done += """
----
-You're an expert software engineer with shell and code knowledge.
-
-Instructions:
-
-    - You should use the provided bash execution, reading and writing file tools to complete objective.
-    - First understand about the project by getting the folder structure (ignoring .git, node_modules, venv, etc.)
-    - Always read relevant files before editing.
-    - Do not provide code snippets unless asked by the user, instead directly add/edit the code.
-    - Do not install new tools/packages before ensuring no such tools/package or an alternative already exists.
-    - Do not use artifacts if you have access to the repository and not asked by the user to provide artifacts/snippets. Directly create/update using shell tools.
-    - Do not use Ctrl-c or Ctrl-z or interrupt commands without asking the user, because often the program don't show any update but they still are running.
-    - Do not use echo to write multi-line files, always use FileEdit tool to update a code.
-    
-Additional instructions:
-    Always run `pwd` if you get any file or directory not found error to make sure you're not lost, or to get absolute cwd.
-
-    Always write production ready, syntactically correct code.
+Important note: as soon as you encounter "The user has chosen to disallow the tool call.", immediately stop doing everything and ask user for the reason.
     """
 
             content.append(types.TextContent(type="text", text=output_or_done))

src/wcgw/client/memory.py

@@ -1,7 +1,8 @@
+import json
 import os
 import re
 import shlex
-from typing import Callable, Optional
+from typing import Any, Callable, Optional
 
 from ..types_ import ContextSave
 
@@ -30,7 +31,11 @@ def format_memory(task_memory: ContextSave, relevant_files: str) -> str:
     return memory_data
 
 
-def save_memory(task_memory: ContextSave, relevant_files: str) -> str:
+def save_memory(
+    task_memory: ContextSave,
+    relevant_files: str,
+    bash_state_dict: Optional[dict[str, Any]] = None,
+) -> str:
     app_dir = get_app_dir_xdg()
     memory_dir = os.path.join(app_dir, "memory")
     os.makedirs(memory_dir, exist_ok=True)
@@ -45,6 +50,12 @@ def save_memory(task_memory: ContextSave, relevant_files: str) -> str:
     with open(memory_file_full, "w") as f:
         f.write(memory_data)
 
+    # Save bash state if provided
+    if bash_state_dict is not None:
+        state_file = os.path.join(memory_dir, f"{task_id}_bash_state.json")
+        with open(state_file, "w") as f:
+            json.dump(bash_state_dict, f, indent=2)
+
     return memory_file_full
 
 
@@ -53,7 +64,7 @@ def load_memory[T](
     max_tokens: Optional[int],
     encoder: Callable[[str], list[T]],
     decoder: Callable[[list[T]], str],
-) -> tuple[str, str]:
+) -> tuple[str, str, Optional[dict[str, Any]]]:
     app_dir = get_app_dir_xdg()
     memory_dir = os.path.join(app_dir, "memory")
     memory_file = os.path.join(memory_dir, f"{task_id}.txt")
@@ -75,4 +86,12 @@ def load_memory[T](
         parsed_ = shlex.split(matched_path)
         if parsed_ and len(parsed_) == 1:
             project_root_path = parsed_[0]
-    return project_root_path, data
+
+    # Try to load bash state if exists
+    state_file = os.path.join(memory_dir, f"{task_id}_bash_state.json")
+    bash_state: Optional[dict[str, Any]] = None
+    if os.path.exists(state_file):
+        with open(state_file) as f:
+            bash_state = json.load(f)
+
+    return project_root_path, data, bash_state