mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
fix(gateway): align content_actions with actual service method names
content_actions.note/dm/say/commit/evidence called write_entry/send/ post_to_channel/git.commit/git.diff(base=)/fetch_branch_for_inspection, none of which existed on JournalService/A2AService/MessagingService/ GitService/WorkspaceService. Live smoke threw AttributeError on every content tool. Add the matching gateway-shaped adapters on each service (scope-string -> JournalEntryType for note; channel-by-slug -> default group -> active session for say; UUID-or-slug recipient resolution for dm; branch-name commit + diff(base=) for commit/evidence; project-aware fetch_branch_for_inspection on workspace). Surfaced live.
This commit is contained in:
@@ -1320,3 +1320,62 @@ class A2AService:
|
|||||||
edited_at=msg.edited_at,
|
edited_at=msg.edited_at,
|
||||||
edit_history=msg.edit_history or [],
|
edit_history=msg.edit_history or [],
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# =========================================================================
|
||||||
|
# GATEWAY (CHOREOGRAPHER + CONTENT_ACTIONS) BACKFILL
|
||||||
|
# =========================================================================
|
||||||
|
|
||||||
|
async def _resolve_slug_from_id(self, agent_id: UUID) -> str:
|
||||||
|
"""Look up an agent's slug from its UUID; raise ValueError if missing."""
|
||||||
|
result = await self.session.execute(
|
||||||
|
select(AgentTable.slug).where(AgentTable.id == agent_id)
|
||||||
|
)
|
||||||
|
slug = result.scalar_one_or_none()
|
||||||
|
if not slug:
|
||||||
|
raise ValueError(f"Agent not found for id {agent_id}")
|
||||||
|
return str(slug)
|
||||||
|
|
||||||
|
async def send(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
from_agent: UUID,
|
||||||
|
to_agent: UUID | str,
|
||||||
|
task_id: UUID,
|
||||||
|
body: str,
|
||||||
|
skill: str | None = None,
|
||||||
|
) -> A2AChatMessage:
|
||||||
|
"""Gateway adapter — send a directed A2A message between two agents.
|
||||||
|
|
||||||
|
Recipient may be either a UUID (choreographer call shape) or a
|
||||||
|
slug string (content_actions call shape). The sender is always a
|
||||||
|
UUID; both ends are resolved to slugs because the
|
||||||
|
conversation/message tables key on slug.
|
||||||
|
|
||||||
|
Resolves to:
|
||||||
|
1. `get_or_create_conversation(sender_slug, recipient_slug, task_id=...)`
|
||||||
|
2. `send_chat_message(conversation.id, sender_slug, content=body, ...)`
|
||||||
|
|
||||||
|
`skill` is recorded in message metadata so the receiver knows which
|
||||||
|
capability is being requested.
|
||||||
|
"""
|
||||||
|
from_slug = await self._resolve_slug_from_id(from_agent)
|
||||||
|
to_slug = (
|
||||||
|
await self._resolve_slug_from_id(to_agent)
|
||||||
|
if isinstance(to_agent, UUID)
|
||||||
|
else to_agent
|
||||||
|
)
|
||||||
|
|
||||||
|
conv = await self.get_or_create_conversation(
|
||||||
|
agent_a=from_slug,
|
||||||
|
agent_b=to_slug,
|
||||||
|
task_id=task_id,
|
||||||
|
)
|
||||||
|
options: dict[str, Any] = {}
|
||||||
|
if skill is not None:
|
||||||
|
options["skill"] = skill
|
||||||
|
return await self.send_chat_message(
|
||||||
|
conversation_id=UUID(conv.id),
|
||||||
|
from_agent=from_slug,
|
||||||
|
content=body,
|
||||||
|
options=options or None,
|
||||||
|
)
|
||||||
|
|||||||
@@ -193,7 +193,7 @@ class ContentActions:
|
|||||||
)
|
)
|
||||||
await self.a2a.send(
|
await self.a2a.send(
|
||||||
from_agent=agent_id,
|
from_agent=agent_id,
|
||||||
to_agent_slug=recipient,
|
to_agent=recipient,
|
||||||
task_id=task_id,
|
task_id=task_id,
|
||||||
body=text,
|
body=text,
|
||||||
skill=skill,
|
skill=skill,
|
||||||
|
|||||||
+87
-10
@@ -1806,21 +1806,98 @@ class GitService(BaseService):
|
|||||||
)
|
)
|
||||||
return str(base_ref)
|
return str(base_ref)
|
||||||
|
|
||||||
async def diff(self, *, branch_name: str) -> str:
|
async def diff(
|
||||||
"""Return the git diff for `branch_name` against its parent or master."""
|
self,
|
||||||
|
*,
|
||||||
|
branch_name: str,
|
||||||
|
base: str | None = None,
|
||||||
|
) -> str:
|
||||||
|
"""Return the git diff for `branch_name` against `base`.
|
||||||
|
|
||||||
|
When `base` is omitted, diffs against the branch's parent (per
|
||||||
|
`parent_branch_for`) which is what the choreographer/PR-review
|
||||||
|
path wants. Content_actions evidence path can pass `HEAD~1` to
|
||||||
|
get just the latest change diff for incremental review.
|
||||||
|
"""
|
||||||
from roboco.services.gateway.merge_chain import parent_branch_for
|
from roboco.services.gateway.merge_chain import parent_branch_for
|
||||||
|
|
||||||
workspace = await self._workspace_for_branch(branch_name)
|
workspace = await self._workspace_for_branch(branch_name)
|
||||||
parent = parent_branch_for(branch_name)
|
if base is None:
|
||||||
# Make sure the parent ref exists locally before diffing.
|
parent = parent_branch_for(branch_name)
|
||||||
await self._run_git(workspace, ["fetch", "origin", parent], check=False)
|
# Make sure the parent ref exists locally before diffing.
|
||||||
diff_result = await self._run_git(
|
await self._run_git(workspace, ["fetch", "origin", parent], check=False)
|
||||||
workspace,
|
diff_args = ["diff", f"origin/{parent}...{branch_name}"]
|
||||||
["diff", f"origin/{parent}...{branch_name}"],
|
else:
|
||||||
check=False,
|
diff_args = ["diff", f"{base}...{branch_name}"]
|
||||||
)
|
diff_result = await self._run_git(workspace, diff_args, check=False)
|
||||||
return diff_result.stdout
|
return diff_result.stdout
|
||||||
|
|
||||||
|
async def commit(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
branch_name: str,
|
||||||
|
message: str,
|
||||||
|
task_id: UUID,
|
||||||
|
files: list[str] | None = None,
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
"""Gateway adapter — commit on `branch_name` with a free-form message.
|
||||||
|
|
||||||
|
Resolves the workspace + project from the branch (same approach as
|
||||||
|
`push_branch`/`create_pr`), stages the requested files (or all),
|
||||||
|
and runs `git commit -m {message}`. Bypasses the conventional-commit
|
||||||
|
template — content_actions has its own validator that asserts the
|
||||||
|
message is descriptive, and the orchestrator-side git template only
|
||||||
|
applies to the structured `commit_for_task` API path.
|
||||||
|
|
||||||
|
Returns a dict shaped for the gateway: ``{"sha": str, "message": str,
|
||||||
|
"files_changed": int, "insertions": int, "deletions": int}``. Tests
|
||||||
|
and downstream gateway code only consume `sha`; the rest is included
|
||||||
|
so we don't have to invent a new shape later.
|
||||||
|
"""
|
||||||
|
workspace = await self._workspace_for_branch(branch_name)
|
||||||
|
await self._assert_on_task_branch(workspace, branch_name)
|
||||||
|
|
||||||
|
# Stage files explicitly when provided; otherwise stage everything
|
||||||
|
# the agent has touched. Mirrors create_commit's staging logic.
|
||||||
|
if files:
|
||||||
|
for file in files:
|
||||||
|
await self._run_git(workspace, ["add", file])
|
||||||
|
else:
|
||||||
|
await self._run_git(workspace, ["add", "-A"])
|
||||||
|
|
||||||
|
# Free-form gateway commit — message is passed verbatim. The
|
||||||
|
# gateway's commit_validator already rejected garbage messages
|
||||||
|
# before we got here; we don't double-validate.
|
||||||
|
await self._run_git(workspace, ["commit", "-m", message])
|
||||||
|
|
||||||
|
log_result = await self._run_git(workspace, ["log", "-1", "--format=%H|%s"])
|
||||||
|
parts = log_result.stdout.strip().split("|", 1)
|
||||||
|
commit_hash = parts[0] if parts else "unknown"
|
||||||
|
full_message = parts[1] if len(parts) > 1 else message
|
||||||
|
|
||||||
|
stat_result = await self._run_git(
|
||||||
|
workspace, ["diff", "--stat", "HEAD~1..HEAD"], check=False
|
||||||
|
)
|
||||||
|
insertions, deletions, files_changed = self._parse_commit_stats(
|
||||||
|
stat_result.stdout
|
||||||
|
)
|
||||||
|
|
||||||
|
# Best-effort link to the task; mirrors commit_for_task. A linking
|
||||||
|
# failure must not lose the commit.
|
||||||
|
task = await self._task_for_branch(branch_name)
|
||||||
|
if task is not None and task.assigned_to is not None:
|
||||||
|
await self._link_commit_to_task(
|
||||||
|
task_id, commit_hash, message, UUID(str(task.assigned_to))
|
||||||
|
)
|
||||||
|
|
||||||
|
return {
|
||||||
|
"sha": commit_hash,
|
||||||
|
"message": full_message,
|
||||||
|
"files_changed": files_changed,
|
||||||
|
"insertions": insertions,
|
||||||
|
"deletions": deletions,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def get_git_service(session: AsyncSession) -> GitService:
|
def get_git_service(session: AsyncSession) -> GitService:
|
||||||
"""Factory function to get git service."""
|
"""Factory function to get git service."""
|
||||||
|
|||||||
@@ -794,6 +794,57 @@ class JournalService(BaseService):
|
|||||||
)
|
)
|
||||||
return await self.add_struggle(agent_id, params)
|
return await self.add_struggle(agent_id, params)
|
||||||
|
|
||||||
|
# Mapping from gateway scope strings (note/decision/reflect/learning/struggle)
|
||||||
|
# to canonical JournalEntryType enum values. Defined as a class-level constant
|
||||||
|
# so the lookup is a single dict access per call.
|
||||||
|
_SCOPE_TO_TYPE: ClassVar[dict[str, JournalEntryType]] = {
|
||||||
|
"note": JournalEntryType.GENERAL,
|
||||||
|
"decision": JournalEntryType.DECISION_LOG,
|
||||||
|
"reflect": JournalEntryType.TASK_REFLECTION,
|
||||||
|
"learning": JournalEntryType.LEARNING,
|
||||||
|
"struggle": JournalEntryType.STRUGGLE,
|
||||||
|
}
|
||||||
|
|
||||||
|
async def write_entry(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
agent_id: UUID,
|
||||||
|
title: str,
|
||||||
|
content: str,
|
||||||
|
scope: str = "note",
|
||||||
|
task_id: UUID | None = None,
|
||||||
|
) -> JournalEntry | None:
|
||||||
|
"""Gateway adapter — write a journal entry by `scope` string.
|
||||||
|
|
||||||
|
The gateway speaks in scope strings (`note`, `decision`, `reflect`,
|
||||||
|
`learning`, `struggle`) while the service stores entries by
|
||||||
|
`JournalEntryType` enum keyed off a `journal_id` (which the agent
|
||||||
|
doesn't carry). This adapter resolves both: maps scope to the enum,
|
||||||
|
looks up or creates the agent's journal, then delegates to
|
||||||
|
`create_entry(JournalEntryCreate(...))`.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: If `scope` is not one of the supported gateway
|
||||||
|
scopes. Caller (gateway) validates the scope set before
|
||||||
|
reaching here, but the guard is kept defensive.
|
||||||
|
"""
|
||||||
|
entry_type = self._SCOPE_TO_TYPE.get(scope)
|
||||||
|
if entry_type is None:
|
||||||
|
raise ValueError(
|
||||||
|
f"unknown scope {scope!r}; "
|
||||||
|
f"expected one of {sorted(self._SCOPE_TO_TYPE)}"
|
||||||
|
)
|
||||||
|
journal = await self.get_or_create_journal(agent_id)
|
||||||
|
return await self.create_entry(
|
||||||
|
JournalEntryCreate(
|
||||||
|
journal_id=journal.id,
|
||||||
|
type=entry_type,
|
||||||
|
title=title,
|
||||||
|
content=content,
|
||||||
|
task_id=task_id,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def get_journal_service(db: AsyncSession) -> JournalService:
|
def get_journal_service(db: AsyncSession) -> JournalService:
|
||||||
"""Factory function for JournalService."""
|
"""Factory function for JournalService."""
|
||||||
|
|||||||
@@ -1753,6 +1753,68 @@ class MessagingService(BaseService):
|
|||||||
)
|
)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
# =========================================================================
|
||||||
|
# GATEWAY (CONTENT_ACTIONS) BACKFILL
|
||||||
|
# =========================================================================
|
||||||
|
|
||||||
|
async def _default_group_for_channel(
|
||||||
|
self,
|
||||||
|
channel: ChannelTable,
|
||||||
|
) -> GroupTable:
|
||||||
|
"""Return a usable group for posting into `channel`.
|
||||||
|
|
||||||
|
Strategy: pick the first existing group ordered by hierarchy_level
|
||||||
|
then name. If the channel has no groups yet (fresh channel), create
|
||||||
|
a single default group. Channels were originally designed to have
|
||||||
|
explicit groups created at provisioning time, but the gateway
|
||||||
|
`say` verb addresses the channel as a whole — so we paper over
|
||||||
|
that boundary here rather than forcing every caller to know about
|
||||||
|
groups.
|
||||||
|
"""
|
||||||
|
groups = await self.list_groups_in_channel(cast("UUID", channel.id))
|
||||||
|
if groups:
|
||||||
|
return groups[0]
|
||||||
|
return await self.create_group(
|
||||||
|
GroupCreateRequest(
|
||||||
|
name="default",
|
||||||
|
channel_id=cast("UUID", channel.id),
|
||||||
|
allowed_roles=[],
|
||||||
|
hierarchy_level=4,
|
||||||
|
members=[],
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
async def post_to_channel(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
agent_id: UUID,
|
||||||
|
channel_slug: str,
|
||||||
|
content: str,
|
||||||
|
task_id: UUID | None = None,
|
||||||
|
) -> MessageTable:
|
||||||
|
"""Gateway adapter — post a message to a channel by slug.
|
||||||
|
|
||||||
|
The gateway `say` verb addresses channels by slug (`backend-cell`,
|
||||||
|
`all-hands`, ...) and doesn't carry session/group IDs. This adapter
|
||||||
|
resolves the channel by slug, picks the channel's default group,
|
||||||
|
gets or creates the active session for that group, then sends a
|
||||||
|
message via `send_message`.
|
||||||
|
|
||||||
|
Channel access is enforced inside `send_message` (channel writers
|
||||||
|
list + role rules); this adapter never bypasses that check.
|
||||||
|
"""
|
||||||
|
channel = await self.get_channel_by_slug_or_raise(channel_slug)
|
||||||
|
group = await self._default_group_for_channel(channel)
|
||||||
|
session = await self.get_or_create_active_session(cast("UUID", group.id))
|
||||||
|
return await self.send_message(
|
||||||
|
MessageCreateRequest(
|
||||||
|
agent_id=agent_id,
|
||||||
|
session_id=cast("UUID", session.id),
|
||||||
|
content=content,
|
||||||
|
task_id=task_id,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
# =============================================================================
|
# =============================================================================
|
||||||
# SERVICE FACTORY
|
# SERVICE FACTORY
|
||||||
|
|||||||
@@ -626,6 +626,110 @@ class WorkspaceService:
|
|||||||
)
|
)
|
||||||
return workspaces
|
return workspaces
|
||||||
|
|
||||||
|
# =========================================================================
|
||||||
|
# GATEWAY (CONTENT_ACTIONS) BACKFILL
|
||||||
|
# =========================================================================
|
||||||
|
|
||||||
|
async def _resolve_branch_to_project_slug(self, branch_name: str) -> str:
|
||||||
|
"""Look up the task that owns `branch_name` and return its project slug.
|
||||||
|
|
||||||
|
Raises WorkspaceError when no task references the branch or the
|
||||||
|
project record is missing — fetching a phantom branch would
|
||||||
|
silently no-op otherwise.
|
||||||
|
"""
|
||||||
|
from sqlalchemy import select
|
||||||
|
|
||||||
|
from roboco.db.tables import TaskTable
|
||||||
|
from roboco.services.project import get_project_service
|
||||||
|
|
||||||
|
result = await self.session.execute(
|
||||||
|
select(TaskTable).where(TaskTable.branch_name == branch_name).limit(1)
|
||||||
|
)
|
||||||
|
task = result.scalar_one_or_none()
|
||||||
|
if task is None:
|
||||||
|
raise WorkspaceError(f"No task references branch {branch_name!r}")
|
||||||
|
project_service = get_project_service(self.session)
|
||||||
|
project = await project_service.get(UUID(str(task.project_id)))
|
||||||
|
if project is None:
|
||||||
|
raise WorkspaceError(
|
||||||
|
f"Task {task.id} for branch {branch_name!r} has no project"
|
||||||
|
)
|
||||||
|
return str(project.slug)
|
||||||
|
|
||||||
|
async def fetch_branch_for_inspection(
|
||||||
|
self,
|
||||||
|
*,
|
||||||
|
agent_id: UUID,
|
||||||
|
branch_name: str,
|
||||||
|
) -> Path:
|
||||||
|
"""Fetch `branch_name` into the inspecting agent's workspace.
|
||||||
|
|
||||||
|
QA / Documenter / PM agents need to read a developer's branch from
|
||||||
|
their own workspace before diffing. This adapter:
|
||||||
|
|
||||||
|
1. Resolves the project from the branch (via the owning task).
|
||||||
|
2. Ensures a healthy workspace for `agent_id` on that project
|
||||||
|
(clones if missing — same path as the agent's first claim).
|
||||||
|
3. Runs `git fetch origin <branch>` with the project token so the
|
||||||
|
branch ref is locally available for `git diff`.
|
||||||
|
|
||||||
|
Returns the workspace path so the caller can chain checkout/diff
|
||||||
|
operations if needed.
|
||||||
|
"""
|
||||||
|
from roboco.services.project import get_project_service
|
||||||
|
|
||||||
|
project_slug = await self._resolve_branch_to_project_slug(branch_name)
|
||||||
|
workspace = await self.ensure_workspace(
|
||||||
|
project_slug=project_slug,
|
||||||
|
agent_id=agent_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
from roboco.utils.crypto import EncryptionError
|
||||||
|
|
||||||
|
project_service = get_project_service(self.session)
|
||||||
|
project = await project_service.get_by_slug(project_slug)
|
||||||
|
git_token: str | None = None
|
||||||
|
if project is not None:
|
||||||
|
try:
|
||||||
|
git_token = await project_service.get_decrypted_token_by_slug(
|
||||||
|
project_slug
|
||||||
|
)
|
||||||
|
except EncryptionError:
|
||||||
|
# Token-decrypt failure (rotated key / corrupted record) is
|
||||||
|
# non-fatal here: a public branch fetch still works without
|
||||||
|
# auth, and a real auth failure surfaces from git below.
|
||||||
|
git_token = None
|
||||||
|
|
||||||
|
prefix: list[str] = []
|
||||||
|
if git_token:
|
||||||
|
import base64
|
||||||
|
|
||||||
|
basic = base64.b64encode(f"x-access-token:{git_token}".encode()).decode()
|
||||||
|
prefix = ["-c", f"http.extraheader=Authorization: Basic {basic}"]
|
||||||
|
|
||||||
|
def _do_fetch() -> subprocess.CompletedProcess[str]:
|
||||||
|
return subprocess.run(
|
||||||
|
["git", *prefix, "fetch", "origin", branch_name],
|
||||||
|
cwd=str(workspace),
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
timeout=settings.workspace_clone_timeout,
|
||||||
|
check=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
result = await asyncio.to_thread(_do_fetch)
|
||||||
|
if result.returncode != 0:
|
||||||
|
logger.warning(
|
||||||
|
"fetch_branch_for_inspection: fetch returned non-zero",
|
||||||
|
branch=branch_name,
|
||||||
|
workspace=str(workspace),
|
||||||
|
stderr=result.stderr.strip(),
|
||||||
|
)
|
||||||
|
# Re-chown so the agent user can still write into .git after our
|
||||||
|
# root-side fetch updated refs/objects.
|
||||||
|
await asyncio.to_thread(_ensure_agent_owned, workspace)
|
||||||
|
return workspace
|
||||||
|
|
||||||
async def delete_workspace(
|
async def delete_workspace(
|
||||||
self,
|
self,
|
||||||
project_slug: str,
|
project_slug: str,
|
||||||
|
|||||||
@@ -306,7 +306,7 @@ async def test_dm_with_active_task_succeeds() -> None:
|
|||||||
assert body["task_id"] == str(task_id)
|
assert body["task_id"] == str(task_id)
|
||||||
a2a_svc.send.assert_awaited_once()
|
a2a_svc.send.assert_awaited_once()
|
||||||
call_kwargs = a2a_svc.send.call_args.kwargs
|
call_kwargs = a2a_svc.send.call_args.kwargs
|
||||||
assert call_kwargs["to_agent_slug"] == "be-qa-1"
|
assert call_kwargs["to_agent"] == "be-qa-1"
|
||||||
assert call_kwargs["skill"] == "code_review"
|
assert call_kwargs["skill"] == "code_review"
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user