mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-29 10:52:26 +00:00
226 lines
8.2 KiB
Python
226 lines
8.2 KiB
Python
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
|
|
|
|
"""Read-oriented, structured Git capability surface for Eigent agents."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
|
|
from camel.toolkits.base import BaseToolkit
|
|
from camel.toolkits.function_tool import FunctionTool
|
|
|
|
from app.agent.toolkit.abstract_toolkit import AbstractToolkit
|
|
from app.utils.listen.toolkit_listen import auto_listen_toolkit
|
|
from app.utils.space_overlay_client import run_context_for_task
|
|
from app.workspace_git import (
|
|
AdvancedGitCommandRejected,
|
|
get_default_advanced_git_service,
|
|
)
|
|
|
|
|
|
@auto_listen_toolkit(BaseToolkit)
|
|
class WorkspaceGitToolkit(BaseToolkit, AbstractToolkit):
|
|
"""Inspect the Run's local Content Repository without invoking a shell.
|
|
|
|
Mutating Git operations remain behind typed file/checkpoint flows or the
|
|
policy-gated Desktop Advanced Git panel. This toolkit can classify an
|
|
advanced argv request, but cannot treat model output as user approval.
|
|
"""
|
|
|
|
agent_name: str = "agent"
|
|
|
|
def __init__(
|
|
self,
|
|
api_task_id: str,
|
|
agent_name: str | None = None,
|
|
timeout: float | None = None,
|
|
) -> None:
|
|
super().__init__(timeout)
|
|
self.api_task_id = api_task_id
|
|
if agent_name is not None:
|
|
self.agent_name = agent_name
|
|
|
|
def git_capabilities(self) -> str:
|
|
"""Return structured Git availability and allowed Agent actions."""
|
|
|
|
context, service, repository = self._scope()
|
|
if repository is None:
|
|
return json.dumps(
|
|
{"available": False, "reason": "content_repository_disabled"}
|
|
)
|
|
probe = service.git.probe(context.working_directory)
|
|
return json.dumps(
|
|
{
|
|
"available": True,
|
|
"repository_id": repository.repository_id,
|
|
"logical_root": ".",
|
|
"current_branch": probe.branch,
|
|
"base_commit": probe.head_oid,
|
|
"version_coverage": repository.version_coverage,
|
|
"allowed_actions": [
|
|
"status",
|
|
"log",
|
|
"branch_list",
|
|
"advanced_preview",
|
|
"checkpoint_via_file_tools",
|
|
"merge_request_via_coordinator",
|
|
],
|
|
"advanced_execution": "requires_user_review_in_desktop",
|
|
"model_guidance": (
|
|
"Do not use Git options that launch editors, hooks, "
|
|
"pagers, filters, helpers, signers, or other external "
|
|
"programs. Request a HumanInteraction before dangerous "
|
|
"or destructive Git work. A preview is never approval."
|
|
),
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
|
|
def git_status(self) -> str:
|
|
"""Return repository health and pending managed-file counts."""
|
|
|
|
_, service, repository = self._scope()
|
|
if repository is None:
|
|
return json.dumps({"available": False})
|
|
status = service.content.status(repository.repository_id)
|
|
return json.dumps(
|
|
{
|
|
"available": True,
|
|
"state": status.repository.state,
|
|
"healthy": status.diagnostics.healthy,
|
|
"issues": list(status.diagnostics.issues),
|
|
"branch": status.diagnostics.state_token.branch_or_detached_head,
|
|
"head_oid": status.diagnostics.state_token.head_oid,
|
|
"repo_state_digest": status.diagnostics.state_token.digest,
|
|
"pending_managed_paths": list(status.pending_managed_paths),
|
|
"pending_managed_paths_truncated": (
|
|
status.pending_managed_paths_truncated
|
|
),
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
|
|
def git_log(self, limit: int = 20) -> str:
|
|
"""Return bounded commit metadata; file content is never embedded.
|
|
|
|
Args:
|
|
limit: Maximum number of recent commits to return, from 1 to 100.
|
|
"""
|
|
|
|
_, service, repository = self._scope()
|
|
if repository is None:
|
|
return json.dumps({"available": False})
|
|
history = service.history(
|
|
repository_id=repository.repository_id,
|
|
limit=max(1, min(limit, 100)),
|
|
)
|
|
return json.dumps(
|
|
{"available": True, "commits": history["commits"]},
|
|
sort_keys=True,
|
|
)
|
|
|
|
def git_branches(self) -> str:
|
|
"""Return active and archived branch metadata for this repository."""
|
|
|
|
_, service, repository = self._scope()
|
|
if repository is None:
|
|
return json.dumps({"available": False})
|
|
history = service.history(repository_id=repository.repository_id)
|
|
return json.dumps(
|
|
{
|
|
"available": True,
|
|
"branches": history["branches"],
|
|
"remotes": history["remotes"],
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
|
|
def advanced_git_preview(
|
|
self,
|
|
argv: list[str],
|
|
operation_request_id: str = "",
|
|
) -> str:
|
|
"""Classify argv without executing it.
|
|
|
|
Use this to explain the required permission class. A write preview is
|
|
not approval and cannot be executed by this read-only Agent surface.
|
|
|
|
Args:
|
|
argv: Git arguments excluding the ``git`` executable, as an array.
|
|
operation_request_id: Stable retry identifier for this preview.
|
|
"""
|
|
|
|
context, service, repository = self._scope()
|
|
if repository is None:
|
|
return json.dumps({"available": False})
|
|
request_id = (
|
|
operation_request_id.strip() or f"agent-preview-{uuid.uuid4().hex}"
|
|
)
|
|
try:
|
|
preview = service.preview(
|
|
space_id=context.space_id,
|
|
repository_id=repository.repository_id,
|
|
argv=tuple(argv),
|
|
operation_request_id=request_id,
|
|
)
|
|
except AdvancedGitCommandRejected as exc:
|
|
return json.dumps(
|
|
{
|
|
"accepted": False,
|
|
"rejection": {
|
|
"code": exc.reason_code,
|
|
"reason": str(exc),
|
|
"remediation": exc.remediation,
|
|
"human_interaction_required": (
|
|
exc.human_interaction_required
|
|
),
|
|
},
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
return json.dumps(
|
|
{
|
|
"accepted": True,
|
|
"classification": preview.classification.operation,
|
|
"safety_class": preview.classification.safety_class.value,
|
|
"external_side_effect": (
|
|
preview.classification.external_side_effect
|
|
),
|
|
"requires_user_confirmation": preview.requires_confirmation,
|
|
"display_argv": list(preview.display_argv),
|
|
"action_digest": preview.action_digest,
|
|
"execution": (
|
|
"request HumanInteraction when confirmation is required; "
|
|
"then open Desktop Version history to review"
|
|
),
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
|
|
def _scope(self):
|
|
context = run_context_for_task(self.api_task_id)
|
|
if context is None:
|
|
raise RuntimeError(
|
|
"WorkspaceGitToolkit requires a durable RunContext"
|
|
)
|
|
service = get_default_advanced_git_service()
|
|
repository = service.journal.get_space_git_repository(
|
|
space_id=context.space_id
|
|
)
|
|
return context, service, repository
|
|
|
|
def get_tools(self) -> list[FunctionTool]:
|
|
return [
|
|
FunctionTool(self.git_capabilities),
|
|
FunctionTool(self.git_status),
|
|
FunctionTool(self.git_log),
|
|
FunctionTool(self.git_branches),
|
|
FunctionTool(self.advanced_git_preview),
|
|
]
|