KINGMAKER\67483
1 week ago
commit
b08c7ddce7
@ -0,0 +1,16 @@
|
||||
.gcode/
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*.egg-info/
|
||||
.venv/
|
||||
venv/
|
||||
dist/
|
||||
build/
|
||||
.env
|
||||
.pytest_cache/
|
||||
.mypy_cache/
|
||||
.idea/
|
||||
*.db
|
||||
*.db-journal
|
||||
*.db-wal
|
||||
*.db-shm
|
||||
@ -0,0 +1,28 @@
|
||||
# G-CODE 模型注册表示例。复制到 %APPDATA%\gcode\models.yaml
|
||||
# 密钥只放环境变量，不要把 Key 写进本文件。
|
||||
default: agicto/deepseek-v4-flash
|
||||
|
||||
providers:
|
||||
deepseek:
|
||||
api: openai_compatible
|
||||
base_url: https://api.deepseek.com/v1
|
||||
api_key_env: DEEPSEEK_API_KEY
|
||||
models:
|
||||
- id: deepseek-chat
|
||||
vision: false
|
||||
reasoning: false
|
||||
- id: deepseek-reasoner
|
||||
vision: false
|
||||
reasoning: true
|
||||
|
||||
agicto:
|
||||
api: openai_compatible
|
||||
base_url: https://api.agicto.cn/v1
|
||||
api_key_env: GCODE_API_KEY
|
||||
models:
|
||||
- id: deepseek-v4-flash
|
||||
vision: false
|
||||
reasoning: false
|
||||
- id: deepseek-v4-pro
|
||||
vision: false
|
||||
reasoning: true
|
||||
@ -0,0 +1,41 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=61.0", "wheel"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "gcode-agent"
|
||||
version = "0.1.0"
|
||||
description = "G-CODE: Windows-first personal coding agent (ReAct + TUI / REPL)"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.10"
|
||||
license = { text = "MIT" }
|
||||
authors = [{ name = "G-CODE" }]
|
||||
|
||||
dependencies = [
|
||||
"langgraph>=0.2.0",
|
||||
"langchain-core>=0.3.0",
|
||||
"langchain-openai>=0.2.0",
|
||||
"langgraph-checkpoint-sqlite>=2.0.0",
|
||||
"aiosqlite>=0.20.0",
|
||||
"pyyaml>=6.0",
|
||||
"textual>=0.80.0",
|
||||
"langchain-mcp-adapters>=0.1.0",
|
||||
"mcp>=1.0.0,<2.0.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = [
|
||||
"pytest>=8.0",
|
||||
"pytest-asyncio>=0.24.0",
|
||||
]
|
||||
|
||||
[project.scripts]
|
||||
gcode = "gcode.cli:main"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["src"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
asyncio_mode = "auto"
|
||||
testpaths = ["tests"]
|
||||
pythonpath = ["src"]
|
||||
@ -0,0 +1,3 @@
|
||||
"""G-CODE: Windows-first personal coding agent."""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
@ -0,0 +1,4 @@
|
||||
from gcode.cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@ -0,0 +1,16 @@
|
||||
"""UI-facing stream events. TUI and REPL both consume these; they never import LangGraph internals."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Literal
|
||||
|
||||
EventType = Literal["thinking", "text", "tool_start", "tool_end", "done", "error"]
|
||||
|
||||
|
||||
@dataclass
|
||||
class AgentEvent:
|
||||
type: EventType
|
||||
text: str = ""
|
||||
name: str = ""
|
||||
extra: dict[str, Any] = field(default_factory=dict)
|
||||
@ -0,0 +1,264 @@
|
||||
"""LangGraph ReAct heart: START → llm_call ⇄ tool_node → END. No global llm/agent."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import operator
|
||||
from collections.abc import AsyncIterator
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
|
||||
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, SystemMessage, ToolMessage
|
||||
from langgraph.graph import END, START, StateGraph
|
||||
from typing_extensions import Annotated, TypedDict
|
||||
|
||||
from gcode.events import AgentEvent
|
||||
from gcode.models.openai_compatible import extract_reasoning, extract_text
|
||||
from gcode.prompt import dynamic_context, static_system_prompt
|
||||
from gcode.session import Session
|
||||
|
||||
MAX_LLM_CALLS = 40
|
||||
|
||||
|
||||
class MessagesState(TypedDict):
|
||||
messages: Annotated[list[AnyMessage], operator.add]
|
||||
llm_calls: int
|
||||
|
||||
|
||||
class Agent:
|
||||
def __init__(
|
||||
self,
|
||||
model: Any,
|
||||
tools: list[Any],
|
||||
*,
|
||||
session: Session | None = None,
|
||||
model_id: str = "",
|
||||
has_browser: bool = False,
|
||||
cwd: Path | None = None,
|
||||
) -> None:
|
||||
self._tools = {t.name: t for t in tools}
|
||||
self._model = model.bind_tools(tools)
|
||||
self._session = session
|
||||
self._model_id = model_id
|
||||
self._has_browser = has_browser
|
||||
self._cwd = cwd or Path.cwd()
|
||||
self._static = static_system_prompt(has_browser=has_browser)
|
||||
checkpointer = session.checkpointer if session else None
|
||||
self._graph = self._build(checkpointer)
|
||||
|
||||
def _build(self, checkpointer: Any) -> Any:
|
||||
builder = StateGraph(MessagesState)
|
||||
builder.add_node("llm_call", self._llm_call)
|
||||
builder.add_node("tool_node", self._tool_node)
|
||||
builder.add_edge(START, "llm_call")
|
||||
builder.add_conditional_edges(
|
||||
"llm_call",
|
||||
self._should_continue,
|
||||
{"tool_node": "tool_node", END: END},
|
||||
)
|
||||
builder.add_edge("tool_node", "llm_call")
|
||||
return builder.compile(checkpointer=checkpointer)
|
||||
|
||||
def _graph_config(self) -> dict[str, Any] | None:
|
||||
if self._session and self._session.checkpointer:
|
||||
return self._session.config()
|
||||
return None
|
||||
|
||||
async def invoke_turn(self, user_text: str) -> list[AnyMessage]:
|
||||
payload: MessagesState = {
|
||||
"messages": [HumanMessage(content=user_text)],
|
||||
"llm_calls": 0,
|
||||
}
|
||||
config = self._graph_config()
|
||||
result = await self._graph.ainvoke(payload, config=config)
|
||||
return list(result["messages"])
|
||||
|
||||
async def stream_turn(self, user_text: str) -> AsyncIterator[AgentEvent]:
|
||||
payload: MessagesState = {
|
||||
"messages": [HumanMessage(content=user_text)],
|
||||
"llm_calls": 0,
|
||||
}
|
||||
config = self._graph_config()
|
||||
try:
|
||||
async for ev in self._graph.astream_events(payload, config=config, version="v2"):
|
||||
mapped = _map_stream_event(ev)
|
||||
if mapped is not None:
|
||||
yield mapped
|
||||
yield AgentEvent(type="done")
|
||||
except Exception as exc:
|
||||
yield AgentEvent(type="error", text=f"{type(exc).__name__}: {exc}")
|
||||
|
||||
async def _llm_call(self, state: MessagesState) -> dict[str, Any]:
|
||||
calls = int(state.get("llm_calls") or 0)
|
||||
if calls >= MAX_LLM_CALLS:
|
||||
return {
|
||||
"messages": [
|
||||
AIMessage(content="已达到本轮最大循环次数。请把任务拆小，或在下一轮继续。")
|
||||
],
|
||||
"llm_calls": calls,
|
||||
}
|
||||
|
||||
dyn = dynamic_context(self._cwd, self._model_id)
|
||||
messages = _assemble_messages(self._static, dyn, list(state.get("messages") or []))
|
||||
messages = _strip_images(messages)
|
||||
messages = _repair_incomplete_tool_calls(messages)
|
||||
|
||||
try:
|
||||
response = await self._model.ainvoke(messages)
|
||||
return {"messages": [response], "llm_calls": calls + 1}
|
||||
except Exception as exc:
|
||||
err = str(exc)
|
||||
if "tool_calls" in err and "tool_call_id" in err:
|
||||
try:
|
||||
repaired = _repair_incomplete_tool_calls(messages)
|
||||
response = await self._model.ainvoke(repaired)
|
||||
return {"messages": [response], "llm_calls": calls + 1}
|
||||
except Exception:
|
||||
pass
|
||||
friendly = f"模型请求失败: {type(exc).__name__}: {err}"
|
||||
return {"messages": [AIMessage(content=friendly)], "llm_calls": calls + 1}
|
||||
|
||||
async def _tool_node(self, state: MessagesState) -> dict[str, Any]:
|
||||
import asyncio
|
||||
|
||||
last = state["messages"][-1]
|
||||
tool_calls = getattr(last, "tool_calls", None) or []
|
||||
|
||||
async def one(tc: dict[str, Any]) -> ToolMessage:
|
||||
name = tc.get("name") or ""
|
||||
call_id = tc.get("id") or ""
|
||||
args = tc.get("args") or {}
|
||||
tool = self._tools.get(name)
|
||||
if tool is None:
|
||||
available = ", ".join(sorted(self._tools))
|
||||
return ToolMessage(
|
||||
content=f"工具 '{name}' 不存在。可用: {available}",
|
||||
tool_call_id=call_id,
|
||||
name=name,
|
||||
)
|
||||
try:
|
||||
observation = await tool.ainvoke(args)
|
||||
except Exception as exc:
|
||||
observation = f"工具 '{name}' 调用失败: {type(exc).__name__}: {exc}"
|
||||
if not isinstance(observation, str):
|
||||
observation = str(observation)
|
||||
return ToolMessage(content=observation, tool_call_id=call_id, name=name)
|
||||
|
||||
if len(tool_calls) <= 1:
|
||||
results = [await one(tool_calls[0])] if tool_calls else []
|
||||
else:
|
||||
results = list(await asyncio.gather(*[one(tc) for tc in tool_calls]))
|
||||
return {"messages": results}
|
||||
|
||||
def _should_continue(self, state: MessagesState) -> Literal["tool_node", "__end__"]:
|
||||
messages = state.get("messages") or []
|
||||
if not messages:
|
||||
return END
|
||||
last = messages[-1]
|
||||
if getattr(last, "tool_calls", None):
|
||||
return "tool_node"
|
||||
return END
|
||||
|
||||
|
||||
def _assemble_messages(static: str, dyn: str, history: list[AnyMessage]) -> list[AnyMessage]:
|
||||
out: list[AnyMessage] = [SystemMessage(content=static)]
|
||||
if not history:
|
||||
out.append(HumanMessage(content=dyn))
|
||||
return out
|
||||
|
||||
first = history[0]
|
||||
if isinstance(first, HumanMessage) and isinstance(first.content, str):
|
||||
out.append(HumanMessage(content=f"{dyn}\n\n---\n\n{first.content}"))
|
||||
out.extend(history[1:])
|
||||
return out
|
||||
|
||||
out.append(HumanMessage(content=dyn))
|
||||
out.extend(history)
|
||||
return out
|
||||
|
||||
|
||||
def _strip_images(messages: list[AnyMessage]) -> list[AnyMessage]:
|
||||
cleaned: list[AnyMessage] = []
|
||||
for msg in messages:
|
||||
if isinstance(msg, HumanMessage) and isinstance(msg.content, list):
|
||||
texts = [
|
||||
part.get("text", "")
|
||||
for part in msg.content
|
||||
if isinstance(part, dict) and part.get("type") == "text"
|
||||
]
|
||||
if texts:
|
||||
cleaned.append(HumanMessage(content="\n".join(texts)))
|
||||
continue
|
||||
cleaned.append(msg)
|
||||
return cleaned
|
||||
|
||||
|
||||
def _repair_incomplete_tool_calls(messages: list[AnyMessage]) -> list[AnyMessage]:
|
||||
declared: set[str] = set()
|
||||
for msg in messages:
|
||||
if isinstance(msg, AIMessage) and getattr(msg, "tool_calls", None):
|
||||
for tc in msg.tool_calls:
|
||||
if tc.get("id"):
|
||||
declared.add(tc["id"])
|
||||
if not declared:
|
||||
return messages
|
||||
|
||||
responded: set[str] = set()
|
||||
for msg in messages:
|
||||
if isinstance(msg, ToolMessage) and msg.tool_call_id:
|
||||
responded.add(msg.tool_call_id)
|
||||
|
||||
missing = declared - responded
|
||||
if not missing:
|
||||
return messages
|
||||
|
||||
repaired: list[AnyMessage] = []
|
||||
for msg in messages:
|
||||
repaired.append(msg)
|
||||
if isinstance(msg, AIMessage) and getattr(msg, "tool_calls", None):
|
||||
for tc in msg.tool_calls:
|
||||
if tc.get("id") in missing:
|
||||
repaired.append(
|
||||
ToolMessage(
|
||||
content=f"[会话恢复] 工具 '{tc.get('name')}' 在上次对话中被中断，已跳过。",
|
||||
tool_call_id=tc["id"],
|
||||
name=tc.get("name") or "",
|
||||
)
|
||||
)
|
||||
return repaired
|
||||
|
||||
|
||||
def _map_stream_event(ev: dict[str, Any]) -> AgentEvent | None:
|
||||
kind = ev.get("event") or ""
|
||||
meta = ev.get("metadata") or {}
|
||||
node = meta.get("langgraph_node")
|
||||
data = ev.get("data") or {}
|
||||
|
||||
if kind == "on_chat_model_stream" and node == "llm_call":
|
||||
chunk = data.get("chunk")
|
||||
if chunk is None:
|
||||
return None
|
||||
thinking = extract_reasoning(chunk)
|
||||
text = extract_text(chunk)
|
||||
if thinking and not text:
|
||||
return AgentEvent(type="thinking", text=thinking)
|
||||
if thinking and text:
|
||||
return AgentEvent(type="text", text=text, extra={"thinking": thinking})
|
||||
if text:
|
||||
return AgentEvent(type="text", text=text)
|
||||
if thinking:
|
||||
return AgentEvent(type="thinking", text=thinking)
|
||||
return None
|
||||
|
||||
if kind == "on_tool_start":
|
||||
name = ev.get("name") or ""
|
||||
return AgentEvent(type="tool_start", name=name, extra={"input": data.get("input")})
|
||||
|
||||
if kind == "on_tool_end":
|
||||
name = ev.get("name") or ""
|
||||
output = data.get("output")
|
||||
text = output if isinstance(output, str) else str(output or "")
|
||||
if len(text) > 2000:
|
||||
text = text[:2000] + "..."
|
||||
return AgentEvent(type="tool_end", name=name, text=text)
|
||||
|
||||
return None
|
||||
@ -0,0 +1,3 @@
|
||||
from gcode.mcp.client import PlaywrightMCP, ensure_mcp_config
|
||||
|
||||
__all__ = ["PlaywrightMCP", "ensure_mcp_config"]
|
||||
@ -0,0 +1,4 @@
|
||||
from gcode.models.openai_compatible import create_chat_model
|
||||
from gcode.models.registry import ModelSpec, load_registry, resolve_model
|
||||
|
||||
__all__ = ["ModelSpec", "load_registry", "resolve_model", "create_chat_model"]
|
||||
@ -0,0 +1,95 @@
|
||||
"""OpenAI-compatible Chat model. Subclass locally so reasoning_content is not a global LangChain patch."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from langchain_openai import ChatOpenAI
|
||||
|
||||
from gcode.models.registry import ModelSpec
|
||||
|
||||
|
||||
class OpenAICompatibleChat(ChatOpenAI):
|
||||
"""ChatOpenAI plus optional reasoning_content round-trip for compatible gateways."""
|
||||
|
||||
def _get_request_payload(self, input_: Any, **kwargs: Any) -> dict[str, Any]:
|
||||
payload = super()._get_request_payload(input_, **kwargs)
|
||||
try:
|
||||
_attach_reasoning(payload, input_)
|
||||
except Exception:
|
||||
pass
|
||||
return payload
|
||||
|
||||
|
||||
def create_chat_model(spec: ModelSpec) -> OpenAICompatibleChat:
|
||||
if not spec.api_key:
|
||||
raise RuntimeError(
|
||||
f"未找到 API Key。请设置环境变量 {spec.api_key_env} "
|
||||
f"（当前模型 {spec.qualified_id}）。"
|
||||
)
|
||||
kwargs: dict[str, Any] = {
|
||||
"model": spec.model_id,
|
||||
"api_key": spec.api_key,
|
||||
"base_url": spec.base_url,
|
||||
"temperature": 0.2,
|
||||
"streaming": True,
|
||||
"timeout": 120,
|
||||
}
|
||||
if spec.extra_body:
|
||||
kwargs["extra_body"] = spec.extra_body
|
||||
return OpenAICompatibleChat(**kwargs)
|
||||
|
||||
|
||||
def extract_reasoning(message: Any) -> str:
|
||||
extra = getattr(message, "additional_kwargs", None) or {}
|
||||
for key in ("reasoning_content", "reasoning"):
|
||||
value = extra.get(key)
|
||||
if isinstance(value, str) and value:
|
||||
return value
|
||||
content = getattr(message, "content", None)
|
||||
if isinstance(content, list):
|
||||
parts: list[str] = []
|
||||
for block in content:
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
btype = block.get("type")
|
||||
if btype in {"reasoning_content", "reasoning", "thinking"}:
|
||||
parts.append(str(block.get("text") or block.get(btype) or ""))
|
||||
if parts:
|
||||
return "".join(parts)
|
||||
return ""
|
||||
|
||||
|
||||
def extract_text(message: Any) -> str:
|
||||
content = getattr(message, "content", None)
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
if isinstance(content, list):
|
||||
parts: list[str] = []
|
||||
for block in content:
|
||||
if isinstance(block, str):
|
||||
parts.append(block)
|
||||
elif isinstance(block, dict) and block.get("type") == "text":
|
||||
parts.append(str(block.get("text") or ""))
|
||||
return "".join(parts)
|
||||
return "" if content is None else str(content)
|
||||
|
||||
|
||||
def _attach_reasoning(payload: dict[str, Any], input_: Any) -> None:
|
||||
messages = payload.get("messages")
|
||||
if not isinstance(messages, list):
|
||||
return
|
||||
converted: list[Any]
|
||||
try:
|
||||
converted = list(input_)
|
||||
except TypeError:
|
||||
return
|
||||
for i, msg in enumerate(converted):
|
||||
if i >= len(messages):
|
||||
break
|
||||
slot = messages[i]
|
||||
if not isinstance(slot, dict) or slot.get("role") != "assistant":
|
||||
continue
|
||||
reasoning = extract_reasoning(msg)
|
||||
if reasoning:
|
||||
slot["reasoning_content"] = reasoning
|
||||
@ -0,0 +1,142 @@
|
||||
"""YAML model registry. Adding a provider = editing models.yaml, not Python."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
from gcode.config import ensure_home, models_yaml_path
|
||||
|
||||
DEFAULT_MODELS_YAML = """# G-CODE 模型注册表。密钥只放环境变量，不要把 Key 写进本文件。
|
||||
default: agicto/deepseek-v4-flash
|
||||
|
||||
providers:
|
||||
deepseek:
|
||||
api: openai_compatible
|
||||
base_url: https://api.deepseek.com/v1
|
||||
api_key_env: DEEPSEEK_API_KEY
|
||||
models:
|
||||
- id: deepseek-chat
|
||||
vision: false
|
||||
reasoning: false
|
||||
- id: deepseek-reasoner
|
||||
vision: false
|
||||
reasoning: true
|
||||
|
||||
agicto:
|
||||
api: openai_compatible
|
||||
base_url: https://api.agicto.cn/v1
|
||||
api_key_env: GCODE_API_KEY
|
||||
models:
|
||||
- id: deepseek-v4-flash
|
||||
vision: false
|
||||
reasoning: false
|
||||
- id: deepseek-v4-pro
|
||||
vision: false
|
||||
reasoning: true
|
||||
"""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ModelSpec:
|
||||
provider: str
|
||||
model_id: str
|
||||
api: str
|
||||
base_url: str
|
||||
api_key_env: str
|
||||
api_key: str
|
||||
vision: bool = False
|
||||
reasoning: bool = False
|
||||
extra_body: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@property
|
||||
def qualified_id(self) -> str:
|
||||
return f"{self.provider}/{self.model_id}"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Registry:
|
||||
default: str
|
||||
models: list[ModelSpec]
|
||||
|
||||
def find(self, query: str) -> ModelSpec:
|
||||
query = query.strip()
|
||||
if not query:
|
||||
raise KeyError("空的模型 id")
|
||||
# provider/model
|
||||
exact = [m for m in self.models if m.qualified_id == query]
|
||||
if exact:
|
||||
return exact[0]
|
||||
by_id = [m for m in self.models if m.model_id == query]
|
||||
if len(by_id) == 1:
|
||||
return by_id[0]
|
||||
if len(by_id) > 1:
|
||||
ids = ", ".join(m.qualified_id for m in by_id)
|
||||
raise KeyError(f"模型 id '{query}' 对应多家，请写成 provider/id。候选: {ids}")
|
||||
raise KeyError(f"未知模型: {query}")
|
||||
|
||||
|
||||
def ensure_models_yaml() -> Path:
|
||||
ensure_home()
|
||||
path = models_yaml_path()
|
||||
if not path.exists():
|
||||
path.write_text(DEFAULT_MODELS_YAML, encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def load_registry(path: Path | None = None) -> Registry:
|
||||
yaml_path = path or ensure_models_yaml()
|
||||
raw = yaml.safe_load(yaml_path.read_text(encoding="utf-8")) or {}
|
||||
providers = raw.get("providers") or {}
|
||||
models: list[ModelSpec] = []
|
||||
for provider, cfg in providers.items():
|
||||
api = (cfg or {}).get("api") or "openai_compatible"
|
||||
if api != "openai_compatible":
|
||||
continue
|
||||
base_url = (cfg or {}).get("base_url") or ""
|
||||
api_key_env = (cfg or {}).get("api_key_env") or "GCODE_API_KEY"
|
||||
api_key = _read_key(api_key_env)
|
||||
for item in (cfg or {}).get("models") or []:
|
||||
mid = item.get("id")
|
||||
if not mid:
|
||||
continue
|
||||
extra = item.get("extra_body") or {}
|
||||
models.append(
|
||||
ModelSpec(
|
||||
provider=str(provider),
|
||||
model_id=str(mid),
|
||||
api=api,
|
||||
base_url=str(base_url).rstrip("/"),
|
||||
api_key_env=api_key_env,
|
||||
api_key=api_key,
|
||||
vision=bool(item.get("vision", False)),
|
||||
reasoning=bool(item.get("reasoning", False)),
|
||||
extra_body=dict(extra) if isinstance(extra, dict) else {},
|
||||
)
|
||||
)
|
||||
default = str(raw.get("default") or (models[0].qualified_id if models else ""))
|
||||
return Registry(default=default, models=models)
|
||||
|
||||
|
||||
def resolve_model(query: str | None = None, path: Path | None = None) -> ModelSpec:
|
||||
registry = load_registry(path)
|
||||
if not registry.models:
|
||||
raise RuntimeError(
|
||||
f"models.yaml 中没有 openai_compatible 模型。请编辑 {models_yaml_path()}"
|
||||
)
|
||||
return registry.find(query or registry.default)
|
||||
|
||||
|
||||
def _read_key(env_name: str) -> str:
|
||||
import os
|
||||
|
||||
key = os.environ.get(env_name) or ""
|
||||
if key:
|
||||
return key
|
||||
# last-resort shared env, still never from a file in the repo
|
||||
if env_name != "GCODE_API_KEY":
|
||||
return os.environ.get("GCODE_API_KEY") or ""
|
||||
return ""
|
||||
@ -0,0 +1,64 @@
|
||||
"""HITL gates. UI injects the confirm callback; tools never import the graph."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import difflib
|
||||
from collections.abc import Awaitable, Callable
|
||||
from pathlib import Path
|
||||
|
||||
ConfirmFn = Callable[[str, str], Awaitable[bool]]
|
||||
|
||||
REJECT_WRITE = "用户拒绝本次写入，请改用其它方案，不要重复同一写入。"
|
||||
REJECT_SHELL = "用户拒绝执行该命令，请改用其它方案，不要重复同一命令。"
|
||||
|
||||
|
||||
def unified_diff(path: str, before: str, after: str, n: int = 3) -> str:
|
||||
before_lines = before.splitlines(keepends=True)
|
||||
after_lines = after.splitlines(keepends=True)
|
||||
if before_lines and not before_lines[-1].endswith("\n"):
|
||||
before_lines[-1] += "\n"
|
||||
if after_lines and not after_lines[-1].endswith("\n"):
|
||||
after_lines[-1] += "\n"
|
||||
diff = difflib.unified_diff(
|
||||
before_lines,
|
||||
after_lines,
|
||||
fromfile=f"a/{path}",
|
||||
tofile=f"b/{path}",
|
||||
n=n,
|
||||
)
|
||||
text = "".join(diff)
|
||||
if not text.strip():
|
||||
return "(无差异)"
|
||||
if len(text) > 12_000:
|
||||
return text[:12_000] + "\n... [diff 过长，已截断]"
|
||||
return text
|
||||
|
||||
|
||||
class SafetyGate:
|
||||
"""Serializes confirm prompts so parallel tool_calls don't interleave y/N."""
|
||||
|
||||
def __init__(self, confirm: ConfirmFn | None = None) -> None:
|
||||
self._confirm = confirm
|
||||
self._lock = asyncio.Lock()
|
||||
|
||||
def set_confirm(self, confirm: ConfirmFn) -> None:
|
||||
self._confirm = confirm
|
||||
|
||||
async def allow_write(self, path: str | Path, before: str, after: str) -> bool:
|
||||
path_s = str(path)
|
||||
body = f"即将写入文件: {path_s}\n\n{unified_diff(path_s, before, after)}"
|
||||
return await self._ask("write", body)
|
||||
|
||||
async def allow_shell(self, command: str, cwd: str | Path) -> bool:
|
||||
body = f"即将执行命令\n工作目录: {cwd}\n\n{command}"
|
||||
return await self._ask("shell", body)
|
||||
|
||||
async def _ask(self, kind: str, detail: str) -> bool:
|
||||
if self._confirm is None:
|
||||
return False
|
||||
async with self._lock:
|
||||
try:
|
||||
return bool(await self._confirm(kind, detail))
|
||||
except Exception:
|
||||
return False
|
||||
@ -0,0 +1,70 @@
|
||||
"""SQLite checkpointer + thread_id. Project data lives in <cwd>/.gcode/."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import aiosqlite
|
||||
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
|
||||
|
||||
from gcode.config import checkpoints_db, ensure_project_dir, session_file
|
||||
|
||||
|
||||
class Session:
|
||||
def __init__(self, thread_id: str | None = None, cwd: Path | None = None) -> None:
|
||||
self.cwd = cwd or Path.cwd()
|
||||
ensure_project_dir(self.cwd)
|
||||
self.thread_id = thread_id or load_thread_id(self.cwd) or "default"
|
||||
self._conn: aiosqlite.Connection | None = None
|
||||
self.checkpointer: AsyncSqliteSaver | None = None
|
||||
save_thread_id(self.thread_id, self.cwd)
|
||||
|
||||
def config(self) -> dict[str, Any]:
|
||||
return {"configurable": {"thread_id": self.thread_id}}
|
||||
|
||||
async def start(self) -> None:
|
||||
db = checkpoints_db(self.cwd)
|
||||
self._conn = await aiosqlite.connect(str(db))
|
||||
await self._conn.execute("PRAGMA journal_mode=WAL")
|
||||
await self._conn.execute("PRAGMA synchronous=FULL")
|
||||
self.checkpointer = AsyncSqliteSaver(self._conn)
|
||||
|
||||
def new_thread(self) -> str:
|
||||
self.thread_id = str(uuid.uuid4())
|
||||
save_thread_id(self.thread_id, self.cwd)
|
||||
return self.thread_id
|
||||
|
||||
async def close(self) -> None:
|
||||
if self._conn is None:
|
||||
return
|
||||
try:
|
||||
await self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
||||
except Exception:
|
||||
pass
|
||||
await self._conn.close()
|
||||
self._conn = None
|
||||
self.checkpointer = None
|
||||
|
||||
|
||||
def load_thread_id(cwd: Path | None = None) -> str | None:
|
||||
path = session_file(cwd)
|
||||
if not path.is_file():
|
||||
return None
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
tid = data.get("thread_id")
|
||||
return str(tid) if tid else None
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def save_thread_id(thread_id: str, cwd: Path | None = None) -> None:
|
||||
path = session_file(cwd)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(
|
||||
json.dumps({"thread_id": thread_id}, ensure_ascii=False, indent=2),
|
||||
encoding="utf-8",
|
||||
)
|
||||
@ -0,0 +1,4 @@
|
||||
from gcode.tools.fs import make_fs_tools, resolve_path
|
||||
from gcode.tools.shell import make_shell_tools
|
||||
|
||||
__all__ = ["make_fs_tools", "make_shell_tools", "resolve_path"]
|
||||
@ -0,0 +1,3 @@
|
||||
from gcode.ui.detect import detect_terminal, resolve_ui_mode
|
||||
|
||||
__all__ = ["detect_terminal", "resolve_ui_mode"]
|
||||
@ -0,0 +1,37 @@
|
||||
"""Open %EDITOR% or Notepad, then return the saved text as user input."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def capture_from_editor() -> str:
|
||||
editor = (os.environ.get("EDITOR") or os.environ.get("VISUAL") or "").strip()
|
||||
suffix = ".md"
|
||||
fd, raw = tempfile.mkstemp(prefix="gcode-edit-", suffix=suffix)
|
||||
os.close(fd)
|
||||
path = Path(raw)
|
||||
try:
|
||||
path.write_text("", encoding="utf-8")
|
||||
argv = _editor_argv(editor, path)
|
||||
subprocess.run(argv, check=False)
|
||||
return path.read_text(encoding="utf-8")
|
||||
finally:
|
||||
try:
|
||||
path.unlink(missing_ok=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _editor_argv(editor: str, path: Path) -> list[str]:
|
||||
if editor:
|
||||
if sys.platform == "win32" and editor.lower() in {"code", "code.exe", "cursor", "cursor.exe"}:
|
||||
return [editor, "--wait", str(path)]
|
||||
return [editor, str(path)]
|
||||
if sys.platform == "win32":
|
||||
return ["notepad.exe", str(path)]
|
||||
return ["vi", str(path)]
|
||||
@ -0,0 +1,3 @@
|
||||
from gcode.ui.tui.app import GCodeApp
|
||||
|
||||
__all__ = ["GCodeApp"]
|
||||
@ -0,0 +1,300 @@
|
||||
"""G-CODE Textual TUI. Own layout and CSS; consumes Agent.stream_turn only."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.binding import Binding
|
||||
from textual.containers import Horizontal, Vertical, VerticalScroll
|
||||
from textual.widgets import Header, Input, Label, RichLog, Static
|
||||
|
||||
from gcode.models.registry import load_registry
|
||||
from gcode.ui.editor import capture_from_editor
|
||||
from gcode.ui.tui.screens import ConfirmScreen, ModelScreen
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from gcode.cli import Runtime
|
||||
|
||||
KITTY_POP = "\x1b[<u"
|
||||
|
||||
HELP_TEXT = (
|
||||
"/help 帮助 | /edit 外部编辑器 | /clear 新会话 | /model 选模型 | /quit 退出"
|
||||
)
|
||||
|
||||
|
||||
class GCodeApp(App[None]):
|
||||
ENABLE_COMMAND_PALETTE = False
|
||||
TITLE = "G-CODE"
|
||||
CSS = """
|
||||
Screen {
|
||||
background: #101418;
|
||||
color: #e6edf3;
|
||||
}
|
||||
Header {
|
||||
background: #1b222c;
|
||||
color: #f0c14b;
|
||||
text-style: bold;
|
||||
}
|
||||
#body {
|
||||
height: 1fr;
|
||||
}
|
||||
#messages {
|
||||
width: 3fr;
|
||||
padding: 1 2;
|
||||
border: tall #2a3340;
|
||||
}
|
||||
#side {
|
||||
width: 1fr;
|
||||
padding: 0 1;
|
||||
background: #141a21;
|
||||
border: tall #2a3340;
|
||||
}
|
||||
#side Label {
|
||||
color: #8b9bb4;
|
||||
padding: 1 0 0 0;
|
||||
}
|
||||
#thinking {
|
||||
height: 1fr;
|
||||
min-height: 6;
|
||||
background: #0c1014;
|
||||
color: #9aa7b5;
|
||||
}
|
||||
#tools {
|
||||
height: 1fr;
|
||||
min-height: 6;
|
||||
}
|
||||
.bubble-user {
|
||||
background: #1e3a5f;
|
||||
color: #dce8f5;
|
||||
padding: 1 2;
|
||||
margin: 0 0 1 8;
|
||||
border: tall #2d5a8c;
|
||||
}
|
||||
.bubble-bot {
|
||||
background: #1a2420;
|
||||
color: #d7e6d7;
|
||||
padding: 1 2;
|
||||
margin: 0 8 1 0;
|
||||
border: tall #3d6b4f;
|
||||
}
|
||||
.bubble-sys {
|
||||
color: #8b9bb4;
|
||||
padding: 0 1 1 1;
|
||||
}
|
||||
.tool-card {
|
||||
background: #1c2530;
|
||||
padding: 0 1;
|
||||
margin: 0 0 1 0;
|
||||
border: tall #3d4d63;
|
||||
color: #c5d0dc;
|
||||
}
|
||||
.tool-card.done {
|
||||
border: tall #3d6b4f;
|
||||
}
|
||||
#status {
|
||||
background: #1b222c;
|
||||
color: #8b9bb4;
|
||||
padding: 0 2;
|
||||
height: 1;
|
||||
}
|
||||
#prompt {
|
||||
dock: bottom;
|
||||
background: #1b222c;
|
||||
border: tall #f0c14b;
|
||||
margin: 0 1 1 1;
|
||||
}
|
||||
#confirm-box, #model-box {
|
||||
background: #1b222c;
|
||||
border: tall #f0c14b;
|
||||
padding: 1 2;
|
||||
width: 80%;
|
||||
max-width: 120;
|
||||
height: auto;
|
||||
max-height: 80%;
|
||||
margin: 4 8;
|
||||
}
|
||||
#confirm-title {
|
||||
text-style: bold;
|
||||
color: #f0c14b;
|
||||
padding-bottom: 1;
|
||||
}
|
||||
#confirm-body {
|
||||
max-height: 24;
|
||||
padding-bottom: 1;
|
||||
}
|
||||
#confirm-buttons {
|
||||
height: auto;
|
||||
align: center middle;
|
||||
}
|
||||
"""
|
||||
|
||||
BINDINGS = [
|
||||
Binding("ctrl+q", "quit", "退出"),
|
||||
Binding("ctrl+c", "quit", "退出", show=False),
|
||||
]
|
||||
|
||||
def __init__(self, rt: Runtime) -> None:
|
||||
super().__init__()
|
||||
self.rt = rt
|
||||
self._bot: Static | None = None
|
||||
self._bot_buf = ""
|
||||
self._busy = False
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Header(show_clock=False)
|
||||
with Horizontal(id="body"):
|
||||
yield VerticalScroll(id="messages")
|
||||
with Vertical(id="side"):
|
||||
yield Label("思考")
|
||||
yield RichLog(id="thinking", wrap=True, highlight=False, markup=False)
|
||||
yield Label("工具")
|
||||
yield VerticalScroll(id="tools")
|
||||
yield Static(self._status_line(), id="status")
|
||||
yield Input(placeholder="输入任务，/help 查看命令", id="prompt")
|
||||
|
||||
def _status_line(self) -> str:
|
||||
mcp = "browser on" if self.rt.mcp.available else "browser off"
|
||||
return (
|
||||
f" {self.rt.spec.qualified_id} | session {self.rt.session.thread_id[:8]} "
|
||||
f"| {mcp} | {HELP_TEXT}"
|
||||
)
|
||||
|
||||
def _refresh_status(self) -> None:
|
||||
self.query_one("#status", Static).update(self._status_line())
|
||||
|
||||
async def on_mount(self) -> None:
|
||||
self.rt.safety.set_confirm(self.confirm)
|
||||
self._disable_kitty()
|
||||
self.set_interval(0.08, self._flush_bot)
|
||||
msgs = self.query_one("#messages", VerticalScroll)
|
||||
await msgs.mount(Static("G-CODE 已就绪。用中文下任务即可。", classes="bubble-sys"))
|
||||
if self.rt.mcp.message:
|
||||
await msgs.mount(Static(self.rt.mcp.message, classes="bubble-sys"))
|
||||
self.query_one(Input).focus()
|
||||
self.sub_title = self.rt.spec.qualified_id
|
||||
|
||||
def _disable_kitty(self) -> None:
|
||||
driver = getattr(self, "_driver", None)
|
||||
write = getattr(driver, "write", None)
|
||||
if callable(write):
|
||||
try:
|
||||
write(KITTY_POP)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
async def confirm(self, kind: str, detail: str) -> bool:
|
||||
title = "确认写入" if kind == "write" else "确认执行命令"
|
||||
result = await self.push_screen_wait(ConfirmScreen(title, detail))
|
||||
return bool(result)
|
||||
|
||||
def _flush_bot(self) -> None:
|
||||
if self._bot is not None:
|
||||
self._bot.update(self._bot_buf or "...")
|
||||
|
||||
async def on_input_submitted(self, event: Input.Submitted) -> None:
|
||||
text = (event.value or "").strip()
|
||||
event.input.value = ""
|
||||
if not text or self._busy:
|
||||
return
|
||||
if text.startswith("/"):
|
||||
await self._command(text)
|
||||
return
|
||||
self.run_worker(self._run_turn(text), exclusive=True, group="turn")
|
||||
|
||||
async def _command(self, text: str) -> None:
|
||||
parts = text.split(maxsplit=1)
|
||||
cmd = parts[0].lower()
|
||||
arg = parts[1].strip() if len(parts) > 1 else ""
|
||||
msgs = self.query_one("#messages", VerticalScroll)
|
||||
if cmd in {"/quit", "/exit", "/q"}:
|
||||
self.exit()
|
||||
return
|
||||
if cmd == "/help":
|
||||
await msgs.mount(Static(HELP_TEXT, classes="bubble-sys"))
|
||||
return
|
||||
if cmd == "/clear":
|
||||
self.rt.session.new_thread()
|
||||
await msgs.remove_children()
|
||||
self.query_one("#thinking", RichLog).clear()
|
||||
await self.query_one("#tools", VerticalScroll).remove_children()
|
||||
await msgs.mount(Static(f"已新开会话 {self.rt.session.thread_id}", classes="bubble-sys"))
|
||||
self._refresh_status()
|
||||
return
|
||||
if cmd == "/edit":
|
||||
with self.suspend():
|
||||
raw = capture_from_editor()
|
||||
if not raw.strip():
|
||||
await msgs.mount(Static("空内容，已取消 /edit", classes="bubble-sys"))
|
||||
return
|
||||
self.run_worker(self._run_turn(raw), exclusive=True, group="turn")
|
||||
return
|
||||
if cmd == "/model":
|
||||
if arg:
|
||||
try:
|
||||
self.rt.switch_model(arg)
|
||||
self.sub_title = self.rt.spec.qualified_id
|
||||
self._refresh_status()
|
||||
await msgs.mount(Static(f"已切换到 {self.rt.spec.qualified_id}", classes="bubble-sys"))
|
||||
except Exception as exc:
|
||||
await msgs.mount(Static(f"切换失败: {exc}", classes="bubble-sys"))
|
||||
return
|
||||
registry = load_registry()
|
||||
labels = [m.qualified_id for m in registry.models]
|
||||
picked = await self.push_screen_wait(ModelScreen(labels, self.rt.spec.qualified_id))
|
||||
if picked:
|
||||
try:
|
||||
self.rt.switch_model(picked)
|
||||
self.sub_title = self.rt.spec.qualified_id
|
||||
self._refresh_status()
|
||||
await msgs.mount(Static(f"已切换到 {self.rt.spec.qualified_id}", classes="bubble-sys"))
|
||||
except Exception as exc:
|
||||
await msgs.mount(Static(f"切换失败: {exc}", classes="bubble-sys"))
|
||||
return
|
||||
await msgs.mount(Static(f"未知命令 {cmd}，输入 /help", classes="bubble-sys"))
|
||||
|
||||
async def _run_turn(self, text: str) -> None:
|
||||
self._busy = True
|
||||
prompt = self.query_one("#prompt", Input)
|
||||
prompt.disabled = True
|
||||
msgs = self.query_one("#messages", VerticalScroll)
|
||||
tools_view = self.query_one("#tools", VerticalScroll)
|
||||
thinking = self.query_one("#thinking", RichLog)
|
||||
preview = text if len(text) < 2000 else text[:2000] + "..."
|
||||
await msgs.mount(Static(preview, classes="bubble-user"))
|
||||
self._bot_buf = ""
|
||||
self._bot = Static("...", classes="bubble-bot")
|
||||
await msgs.mount(self._bot)
|
||||
cards: dict[str, Static] = {}
|
||||
try:
|
||||
async for ev in self.rt.agent.stream_turn(text):
|
||||
if ev.type == "thinking":
|
||||
thinking.write(ev.text)
|
||||
elif ev.type == "text":
|
||||
extra = (ev.extra or {}).get("thinking")
|
||||
if extra:
|
||||
thinking.write(extra)
|
||||
self._bot_buf += ev.text
|
||||
elif ev.type == "tool_start":
|
||||
card = Static(f"{ev.name} ...", classes="tool-card")
|
||||
cards[ev.name] = card
|
||||
await tools_view.mount(card)
|
||||
elif ev.type == "tool_end":
|
||||
card = cards.get(ev.name)
|
||||
preview_out = (ev.text or "").replace("\n", " ")
|
||||
if len(preview_out) > 120:
|
||||
preview_out = preview_out[:120] + "..."
|
||||
label = f"{ev.name} [done] {preview_out}"
|
||||
if card is not None:
|
||||
card.update(label)
|
||||
card.add_class("done")
|
||||
else:
|
||||
await tools_view.mount(Static(label, classes="tool-card done"))
|
||||
elif ev.type == "error":
|
||||
self._bot_buf += ("\n" + ev.text) if self._bot_buf else ev.text
|
||||
self._flush_bot()
|
||||
finally:
|
||||
self._busy = False
|
||||
prompt.disabled = False
|
||||
prompt.focus()
|
||||
msgs.scroll_end(animate=False)
|
||||
@ -0,0 +1,69 @@
|
||||
"""TUI modal screens: HITL confirm and model picker."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from textual.app import ComposeResult
|
||||
from textual.binding import Binding
|
||||
from textual.containers import Horizontal, Vertical, VerticalScroll
|
||||
from textual.screen import ModalScreen
|
||||
from textual.widgets import Button, Label, Static
|
||||
|
||||
|
||||
class ConfirmScreen(ModalScreen[bool]):
|
||||
BINDINGS = [
|
||||
Binding("y", "yes", "确认", show=False),
|
||||
Binding("n", "no", "拒绝", show=False),
|
||||
Binding("escape", "no", "拒绝", show=False),
|
||||
]
|
||||
|
||||
def __init__(self, title: str, body: str) -> None:
|
||||
super().__init__()
|
||||
self._title = title
|
||||
self._body = body
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
with Vertical(id="confirm-box"):
|
||||
yield Label(self._title, id="confirm-title")
|
||||
with VerticalScroll(id="confirm-body"):
|
||||
yield Static(self._body)
|
||||
with Horizontal(id="confirm-buttons"):
|
||||
yield Button("确认 (y)", id="yes", variant="success")
|
||||
yield Button("拒绝 (n)", id="no", variant="error")
|
||||
|
||||
def on_button_pressed(self, event: Button.Pressed) -> None:
|
||||
self.dismiss(event.button.id == "yes")
|
||||
|
||||
def action_yes(self) -> None:
|
||||
self.dismiss(True)
|
||||
|
||||
def action_no(self) -> None:
|
||||
self.dismiss(False)
|
||||
|
||||
|
||||
class ModelScreen(ModalScreen[str | None]):
|
||||
"""Pick a model; dismiss with qualified id or None."""
|
||||
|
||||
def __init__(self, labels: list[str], current: str) -> None:
|
||||
super().__init__()
|
||||
self._labels = labels
|
||||
self._current = current
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
with Vertical(id="model-box"):
|
||||
yield Label("选择模型", id="confirm-title")
|
||||
with VerticalScroll():
|
||||
for index, name in enumerate(self._labels):
|
||||
mark = " (当前)" if name == self._current else ""
|
||||
yield Button(f"{name}{mark}", id=f"model-{index}")
|
||||
yield Button("取消", id="cancel-model")
|
||||
|
||||
def on_button_pressed(self, event: Button.Pressed) -> None:
|
||||
bid = event.button.id or ""
|
||||
if bid == "cancel-model":
|
||||
self.dismiss(None)
|
||||
return
|
||||
if bid.startswith("model-"):
|
||||
index = int(bid.split("-", 1)[1])
|
||||
self.dismiss(self._labels[index])
|
||||
return
|
||||
self.dismiss(None)
|
||||
@ -0,0 +1,159 @@
|
||||
"""Windows process, encoding, and console helpers.
|
||||
|
||||
Qoze always decoded shell output as UTF-8 and only called terminate() on timeout.
|
||||
G-CODE: UTF-8 then GBK; kill the whole process tree with taskkill /T (or POSIX group).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
|
||||
CREATE_NEW_PROCESS_GROUP = 0x00000200
|
||||
CREATE_NO_WINDOW = 0x08000000
|
||||
|
||||
|
||||
def configure_stdio() -> None:
|
||||
"""Prefer UTF-8 for Python I/O. Does not fix cmd IME; that is UI detect + REPL."""
|
||||
os.environ.setdefault("PYTHONUTF8", "1")
|
||||
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
|
||||
if sys.platform == "win32":
|
||||
try:
|
||||
import ctypes
|
||||
|
||||
ctypes.windll.kernel32.SetConsoleCP(65001)
|
||||
ctypes.windll.kernel32.SetConsoleOutputCP(65001)
|
||||
except Exception:
|
||||
pass
|
||||
for stream in (sys.stdout, sys.stderr):
|
||||
try:
|
||||
stream.reconfigure(encoding="utf-8", errors="replace")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def decode_bytes(data: bytes) -> str:
|
||||
"""Decode command output: try UTF-8, then GBK. Empty bytes → empty string."""
|
||||
if not data:
|
||||
return ""
|
||||
try:
|
||||
return data.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
return data.decode("gbk", errors="replace")
|
||||
|
||||
|
||||
def powershell_argv(command: str) -> list[str]:
|
||||
"""Run `command` via PowerShell -EncodedCommand so Chinese arguments survive."""
|
||||
import base64
|
||||
|
||||
encoded = base64.b64encode(command.encode("utf-16-le")).decode("ascii")
|
||||
return [
|
||||
"powershell.exe",
|
||||
"-NoProfile",
|
||||
"-NonInteractive",
|
||||
"-ExecutionPolicy",
|
||||
"Bypass",
|
||||
"-EncodedCommand",
|
||||
encoded,
|
||||
]
|
||||
|
||||
|
||||
def subprocess_flags() -> int:
|
||||
if sys.platform != "win32":
|
||||
return 0
|
||||
return CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW
|
||||
|
||||
|
||||
def kill_process_tree(pid: int) -> None:
|
||||
"""Kill pid and descendants. Windows: taskkill /T; POSIX: kill process group."""
|
||||
if pid <= 0:
|
||||
return
|
||||
if sys.platform == "win32":
|
||||
subprocess.run(
|
||||
["taskkill", "/T", "/F", "/PID", str(pid)],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
check=False,
|
||||
)
|
||||
return
|
||||
try:
|
||||
os.killpg(os.getpgid(pid), signal.SIGTERM)
|
||||
except ProcessLookupError:
|
||||
return
|
||||
except Exception:
|
||||
try:
|
||||
os.kill(pid, signal.SIGTERM)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def parent_process_name() -> str:
|
||||
"""Best-effort parent executable name (used by terminal detect)."""
|
||||
if sys.platform != "win32":
|
||||
return ""
|
||||
try:
|
||||
return _windows_parent_exe()
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def _windows_parent_exe() -> str:
|
||||
import ctypes
|
||||
from ctypes import wintypes
|
||||
|
||||
TH32CS_SNAPPROCESS = 0x00000002
|
||||
|
||||
class PROCESSENTRY32(ctypes.Structure):
|
||||
_fields_ = [
|
||||
("dwSize", wintypes.DWORD),
|
||||
("cntUsage", wintypes.DWORD),
|
||||
("th32ProcessID", wintypes.DWORD),
|
||||
("th32DefaultHeapID", ctypes.POINTER(ctypes.c_ulong)),
|
||||
("th32ModuleID", wintypes.DWORD),
|
||||
("cntThreads", wintypes.DWORD),
|
||||
("th32ParentProcessID", wintypes.DWORD),
|
||||
("pcPriClassBase", ctypes.c_long),
|
||||
("dwFlags", wintypes.DWORD),
|
||||
("szExeFile", ctypes.c_char * 260),
|
||||
]
|
||||
|
||||
kernel32 = ctypes.windll.kernel32
|
||||
snapshot = kernel32.CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0)
|
||||
if snapshot == -1:
|
||||
return ""
|
||||
try:
|
||||
entry = PROCESSENTRY32()
|
||||
entry.dwSize = ctypes.sizeof(PROCESSENTRY32)
|
||||
current = os.getpid()
|
||||
parent_pid = 0
|
||||
if not kernel32.Process32First(snapshot, ctypes.byref(entry)):
|
||||
return ""
|
||||
while True:
|
||||
if entry.th32ProcessID == current:
|
||||
parent_pid = entry.th32ParentProcessID
|
||||
break
|
||||
if not kernel32.Process32Next(snapshot, ctypes.byref(entry)):
|
||||
break
|
||||
if not parent_pid:
|
||||
return ""
|
||||
if not kernel32.Process32First(snapshot, ctypes.byref(entry)):
|
||||
return ""
|
||||
while True:
|
||||
if entry.th32ProcessID == parent_pid:
|
||||
raw = entry.szExeFile.split(b"\x00", 1)[0]
|
||||
return raw.decode("mbcs", errors="replace")
|
||||
if not kernel32.Process32Next(snapshot, ctypes.byref(entry)):
|
||||
break
|
||||
return ""
|
||||
finally:
|
||||
kernel32.CloseHandle(snapshot)
|
||||
|
||||
|
||||
def truncate_output(text: str, limit: int = 50_000) -> str:
|
||||
if len(text) <= limit:
|
||||
return text
|
||||
head = limit - 200
|
||||
return text[:head] + f"\n... [输出过长，已截断，共 {len(text)} 字符]"
|
||||
@ -0,0 +1,12 @@
|
||||
"""Isolate user-dir lookups from the real %APPDATA% during tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolate_home(tmp_path_factory, monkeypatch):
|
||||
root = tmp_path_factory.mktemp("appdata")
|
||||
monkeypatch.setenv("APPDATA", str(root))
|
||||
monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
|
||||
@ -0,0 +1,40 @@
|
||||
from gcode.ui.detect import CONHOST_HINT, detect_terminal, resolve_ui_mode
|
||||
|
||||
|
||||
def test_windows_terminal_by_session() -> None:
|
||||
info = detect_terminal(env={"WT_SESSION": "abc"}, parent_name="")
|
||||
assert info.tui_ok
|
||||
assert info.host == "windows-terminal"
|
||||
|
||||
|
||||
def test_vscode_term_program() -> None:
|
||||
info = detect_terminal(env={"TERM_PROGRAM": "vscode"}, parent_name="unknown")
|
||||
assert info.tui_ok
|
||||
|
||||
|
||||
def test_conhost_cmd() -> None:
|
||||
info = detect_terminal(env={}, parent_name="cmd.exe")
|
||||
assert not info.tui_ok
|
||||
assert info.host == "conhost"
|
||||
|
||||
|
||||
def test_auto_falls_back_to_repl() -> None:
|
||||
info = detect_terminal(env={}, parent_name="cmd.exe")
|
||||
mode, warning = resolve_ui_mode("auto", info)
|
||||
assert mode == "repl"
|
||||
assert "Windows Terminal" in warning
|
||||
assert warning == CONHOST_HINT
|
||||
|
||||
|
||||
def test_force_tui_on_conhost_refused() -> None:
|
||||
info = detect_terminal(env={}, parent_name="cmd.exe")
|
||||
mode, warning = resolve_ui_mode("tui", info)
|
||||
assert mode == "refuse"
|
||||
assert "repl" in warning.lower() or "Windows Terminal" in warning
|
||||
|
||||
|
||||
def test_force_repl_always() -> None:
|
||||
info = detect_terminal(env={"WT_SESSION": "1"}, parent_name="WindowsTerminal.exe")
|
||||
mode, warning = resolve_ui_mode("repl", info)
|
||||
assert mode == "repl"
|
||||
assert warning == ""
|
||||
@ -0,0 +1,97 @@
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from langchain_core.messages import AIMessage
|
||||
|
||||
from gcode.graph import Agent
|
||||
from gcode.safety import SafetyGate
|
||||
from gcode.tools.fs import make_fs_tools
|
||||
|
||||
|
||||
class ScriptedLLM:
|
||||
def __init__(self, responses: list[AIMessage]) -> None:
|
||||
self.responses = list(responses)
|
||||
self.calls = 0
|
||||
|
||||
def bind_tools(self, tools):
|
||||
return self
|
||||
|
||||
async def ainvoke(self, messages, **kwargs):
|
||||
if self.calls >= len(self.responses):
|
||||
return AIMessage(content="(script exhausted)")
|
||||
msg = self.responses[self.calls]
|
||||
self.calls += 1
|
||||
return msg
|
||||
|
||||
|
||||
async def _always(kind: str, detail: str) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
async def _never(kind: str, detail: str) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_one_tool_call_then_end(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
(tmp_path / "README.md").write_text("hello gcode\n", encoding="utf-8")
|
||||
safety = SafetyGate(_always)
|
||||
tools = make_fs_tools(safety, tmp_path)
|
||||
llm = ScriptedLLM(
|
||||
[
|
||||
AIMessage(
|
||||
content="",
|
||||
tool_calls=[
|
||||
{
|
||||
"name": "read_file",
|
||||
"args": {"path": "README.md"},
|
||||
"id": "call_1",
|
||||
"type": "tool_call",
|
||||
}
|
||||
],
|
||||
),
|
||||
AIMessage(content="README 只有一句 hello gcode。"),
|
||||
]
|
||||
)
|
||||
agent = Agent(llm, tools, cwd=tmp_path)
|
||||
messages = await agent.invoke_turn("请读 README")
|
||||
texts = [getattr(m, "content", "") for m in messages]
|
||||
assert any("hello gcode" in str(t) for t in texts)
|
||||
assert any("README 只有一句" in str(t) for t in texts)
|
||||
assert llm.calls == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_replace_denied_file_unchanged(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
target = tmp_path / "app.py"
|
||||
target.write_text("print(1)\n", encoding="utf-8")
|
||||
safety = SafetyGate(_never)
|
||||
tools = make_fs_tools(safety, tmp_path)
|
||||
llm = ScriptedLLM(
|
||||
[
|
||||
AIMessage(
|
||||
content="",
|
||||
tool_calls=[
|
||||
{
|
||||
"name": "replace_in_file",
|
||||
"args": {
|
||||
"path": "app.py",
|
||||
"old_text": "print(1)",
|
||||
"new_text": "print(2)",
|
||||
},
|
||||
"id": "call_w",
|
||||
"type": "tool_call",
|
||||
}
|
||||
],
|
||||
),
|
||||
AIMessage(content="用户拒绝了写入，我改用其它办法。"),
|
||||
]
|
||||
)
|
||||
agent = Agent(llm, tools, cwd=tmp_path)
|
||||
messages = await agent.invoke_turn("把 print(1) 改成 print(2)")
|
||||
joined = "\n".join(str(getattr(m, "content", "")) for m in messages)
|
||||
assert "用户拒绝本次写入" in joined
|
||||
assert target.read_text(encoding="utf-8") == "print(1)\n"
|
||||
assert llm.calls == 2
|
||||
@ -0,0 +1,46 @@
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from gcode.safety import SafetyGate
|
||||
from gcode.tools.fs import make_fs_tools
|
||||
|
||||
|
||||
def _grep(tmp_path: Path, **kwargs):
|
||||
tools = {t.name: t for t in make_fs_tools(SafetyGate(), tmp_path)}
|
||||
return tools["grep"].invoke(kwargs)
|
||||
|
||||
|
||||
def test_grep_finds_line(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
(tmp_path / "src").mkdir()
|
||||
(tmp_path / "src" / "app.py").write_text("def init_agent():\n return 1\n", encoding="utf-8")
|
||||
out = _grep(tmp_path, query="init_agent")
|
||||
assert "src" in out and "app.py" in out
|
||||
assert ":1:" in out
|
||||
assert "init_agent" in out
|
||||
|
||||
|
||||
def test_grep_skips_git(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
(tmp_path / ".git").mkdir()
|
||||
(tmp_path / ".git" / "HEAD").write_text("init_agent", encoding="utf-8")
|
||||
(tmp_path / "ok.py").write_text("x = 1\n", encoding="utf-8")
|
||||
out = _grep(tmp_path, query="init_agent")
|
||||
assert "No matches" in out
|
||||
assert ".git" not in out
|
||||
|
||||
|
||||
def test_grep_rejects_outside(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
out = _grep(tmp_path, query="foo", path="..")
|
||||
assert "已拒绝" in out or "Error" in out
|
||||
|
||||
|
||||
def test_grep_glob_filters(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
(tmp_path / "a.py").write_text("needle\n", encoding="utf-8")
|
||||
(tmp_path / "a.txt").write_text("needle\n", encoding="utf-8")
|
||||
out = _grep(tmp_path, query="needle", glob="*.py")
|
||||
assert "a.py" in out
|
||||
assert "a.txt" not in out
|
||||
@ -0,0 +1,52 @@
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from gcode.tools.fs import resolve_path
|
||||
|
||||
|
||||
def test_relative_inside_cwd(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
(tmp_path / "a.txt").write_text("x", encoding="utf-8")
|
||||
assert resolve_path("a.txt") == (tmp_path / "a.txt").resolve()
|
||||
|
||||
|
||||
def test_rejects_parent_directory(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
with pytest.raises(ValueError, match="已拒绝"):
|
||||
resolve_path("..")
|
||||
|
||||
|
||||
def test_rejects_absolute_outside(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
outsider = tmp_path.parent / "outside-gcode-test.txt"
|
||||
with pytest.raises(ValueError, match="已拒绝"):
|
||||
resolve_path(str(outsider))
|
||||
|
||||
|
||||
def test_allows_user_home_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from gcode.config import get_home_dir
|
||||
|
||||
proj = tmp_path / "proj"
|
||||
proj.mkdir()
|
||||
monkeypatch.chdir(proj)
|
||||
home = get_home_dir()
|
||||
home.mkdir(parents=True, exist_ok=True)
|
||||
target = home / "note.txt"
|
||||
target.write_text("ok", encoding="utf-8")
|
||||
assert resolve_path(str(target), proj) == target.resolve()
|
||||
|
||||
|
||||
def test_list_dir_skips_git(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from gcode.safety import SafetyGate
|
||||
from gcode.tools.fs import make_fs_tools
|
||||
|
||||
monkeypatch.chdir(tmp_path)
|
||||
(tmp_path / ".git").mkdir()
|
||||
(tmp_path / "src").mkdir()
|
||||
(tmp_path / "readme.txt").write_text("x", encoding="utf-8")
|
||||
tools = {t.name: t for t in make_fs_tools(SafetyGate(), tmp_path)}
|
||||
out = tools["list_dir"].invoke({"path": "."})
|
||||
assert "readme.txt" in out
|
||||
assert "src" in out
|
||||
assert ".git" not in out
|
||||
@ -0,0 +1,47 @@
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from gcode.safety import REJECT_SHELL, REJECT_WRITE, SafetyGate, unified_diff
|
||||
from gcode.tools.fs import make_fs_tools
|
||||
from gcode.tools.shell import make_shell_tools
|
||||
|
||||
|
||||
def test_unified_diff_contains_change() -> None:
|
||||
diff = unified_diff("a.py", "print(1)\n", "print(2)\n")
|
||||
assert "-print(1)" in diff
|
||||
assert "+print(2)" in diff
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unique_replace_and_reject(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
path = tmp_path / "a.py"
|
||||
path.write_text("foo\nfoo\n", encoding="utf-8")
|
||||
|
||||
async def deny(kind: str, detail: str) -> bool:
|
||||
return False
|
||||
|
||||
tools = {t.name: t for t in make_fs_tools(SafetyGate(deny), tmp_path)}
|
||||
out = await tools["replace_in_file"].ainvoke(
|
||||
{"path": "a.py", "old_text": "foo", "new_text": "bar"}
|
||||
)
|
||||
assert "出现了 2 次" in out
|
||||
assert path.read_text(encoding="utf-8") == "foo\nfoo\n"
|
||||
|
||||
path.write_text("only-once\n", encoding="utf-8")
|
||||
out = await tools["replace_in_file"].ainvoke(
|
||||
{"path": "a.py", "old_text": "only-once", "new_text": "twice"}
|
||||
)
|
||||
assert REJECT_WRITE in out
|
||||
assert path.read_text(encoding="utf-8") == "only-once\n"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shell_denied_does_not_run() -> None:
|
||||
async def deny(kind: str, detail: str) -> bool:
|
||||
return False
|
||||
|
||||
tools = {t.name: t for t in make_shell_tools(SafetyGate(deny))}
|
||||
out = await tools["execute_command"].ainvoke({"command": "echo should-not-run"})
|
||||
assert REJECT_SHELL in out
|
||||
@ -0,0 +1,9 @@
|
||||
from pathlib import Path
|
||||
|
||||
from gcode.session import load_thread_id, save_thread_id
|
||||
|
||||
|
||||
def test_thread_id_persists(tmp_path: Path) -> None:
|
||||
save_thread_id("thread-abc", tmp_path)
|
||||
assert load_thread_id(tmp_path) == "thread-abc"
|
||||
assert (tmp_path / ".gcode" / "session.json").is_file()
|
||||
@ -0,0 +1,19 @@
|
||||
from gcode.windows import decode_bytes, truncate_output
|
||||
|
||||
|
||||
def test_decode_utf8() -> None:
|
||||
assert decode_bytes("你好".encode("utf-8")) == "你好"
|
||||
|
||||
|
||||
def test_decode_gbk_fallback() -> None:
|
||||
assert decode_bytes("你好".encode("gbk")) == "你好"
|
||||
|
||||
|
||||
def test_decode_empty() -> None:
|
||||
assert decode_bytes(b"") == ""
|
||||
|
||||
|
||||
def test_truncate() -> None:
|
||||
text = "a" * 100
|
||||
assert truncate_output(text, limit=50).endswith("字符]")
|
||||
assert "a" * 20 == truncate_output("a" * 20, limit=50)
|
||||
|
After Width: | Height: | Size: 83 KiB |
Loading…
Reference in new issue