Files
2026-07-01 17:56:31 +08:00

213 lines
8.7 KiB
Python

"""Helpers for soft-topology collaboration and ownership-contract enforcement."""
from __future__ import annotations
from pathlib import Path
from typing import Any
from opc.core.models import Task
_COLLABORATION_TOOLS = {
"send_dm",
"broadcast_issue",
"start_meeting",
"ask_peer_and_wait",
}
_WRITE_TOOLS = {
"file_write",
"file_edit",
"apply_patch",
"shell_exec",
}
def _normalized_role_list(values: list[Any] | None) -> list[str]:
seen: set[str] = set()
result: list[str] = []
for item in values or []:
role_id = str(item or "").strip()
if not role_id or role_id in seen:
continue
seen.add(role_id)
result.append(role_id)
return result
def collect_dynamic_contact_roles(task: Task | None) -> list[str]:
if task is None:
return []
metadata = dict(getattr(task, "metadata", {}) or {})
ownership_contract = dict(metadata.get("ownership_contract", {}) or {})
work_item_gate = dict(metadata.get("work_item_gate", {}) or {})
dynamic = []
dynamic.extend(_normalized_role_list(ownership_contract.get("allowed_collaboration_targets")))
dynamic.extend(_normalized_role_list(ownership_contract.get("downstream_consumer")))
dynamic.extend(_normalized_role_list(metadata.get("dynamic_allowed_contact_roles")))
reviewer_role = str(work_item_gate.get("reviewer_role", "") or metadata.get("manager_role_id", "") or "").strip()
if reviewer_role:
dynamic.append(reviewer_role)
for handoff in list(metadata.get("handoff_log", []) or []):
if isinstance(handoff, dict):
dynamic.append(str(handoff.get("to", "")).strip())
dynamic.append(str(handoff.get("from", "")).strip())
return _normalized_role_list(dynamic)
def effective_contact_roles(
role_id: str,
*,
task: Task | None = None,
org_engine: Any | None = None,
) -> list[str]:
static_roles: list[str] = []
if org_engine is not None and hasattr(org_engine, "get_allowed_contact_roles"):
getter = getattr(org_engine, "get_allowed_contact_roles")
try:
static_roles = list(getter(role_id, task=task) or [])
except TypeError:
static_roles = list(getter(role_id) or [])
return _normalized_role_list([*static_roles, *collect_dynamic_contact_roles(task)])
def _candidate_paths(tool_name: str, arguments: dict[str, Any]) -> list[Path]:
if tool_name in {"file_write", "file_edit", "file_read", "grep", "glob", "file_search", "list_dir"}:
path = str(arguments.get("path", "") or arguments.get("directory", "") or "").strip()
return [Path(path)] if path else []
if tool_name == "apply_patch":
patch = str(arguments.get("patch", "") or "").splitlines()
paths: list[Path] = []
prefixes = ("*** Add File: ", "*** Update File: ", "*** Delete File: ")
for line in patch:
for prefix in prefixes:
if line.startswith(prefix):
raw = line[len(prefix):].strip()
if raw:
paths.append(Path(raw))
return paths
if tool_name == "shell_exec":
cwd = str(arguments.get("working_directory", "") or "").strip()
return [Path(cwd)] if cwd else []
return []
def _resolved_write_roots(task: Task | None) -> list[Path]:
if task is None:
return []
metadata = dict(getattr(task, "metadata", {}) or {})
ownership_contract = dict(metadata.get("ownership_contract", {}) or {})
scope = str(ownership_contract.get("write_scope", "") or "").strip()
roots: list[Path] = []
if scope and scope not in {"assigned_workspace", "read_only"}:
roots.append(Path(scope))
output_dir = str(metadata.get("target_output_dir", "") or "").strip()
if output_dir:
roots.append(Path(output_dir))
return [path.resolve() for path in roots if str(path).strip()]
def _path_within(candidate: Path, root: Path) -> bool:
try:
return candidate.resolve().is_relative_to(root.resolve())
except AttributeError:
resolved = str(candidate.resolve())
root_value = str(root.resolve())
return resolved == root_value or resolved.startswith(root_value.rstrip("/") + "/")
except FileNotFoundError:
resolved = candidate.expanduser().resolve(strict=False)
root_resolved = root.expanduser().resolve(strict=False)
try:
return resolved.is_relative_to(root_resolved)
except AttributeError:
resolved_value = str(resolved)
root_value = str(root_resolved)
return resolved_value == root_value or resolved_value.startswith(root_value.rstrip("/") + "/")
def ownership_guard_violation(
*,
task: Task | None,
tool_name: str,
arguments: dict[str, Any],
org_engine: Any | None = None,
) -> str | None:
if task is None:
return None
metadata = dict(getattr(task, "metadata", {}) or {})
if str(metadata.get("execution_mode", "") or "").strip() != "company_mode":
return None
ownership_contract = dict(metadata.get("ownership_contract", {}) or {})
if not ownership_contract:
return None
if tool_name in _WRITE_TOOLS:
write_scope = str(ownership_contract.get("write_scope", "") or "").strip()
if write_scope == "read_only":
return (
"Ownership contract blocks write-side effects for this work item. "
"This work item is read_only and must not modify files or run mutating shell commands."
)
roots = _resolved_write_roots(task)
candidate_paths = _candidate_paths(tool_name, arguments)
if roots and candidate_paths:
out_of_scope = [str(path) for path in candidate_paths if not any(_path_within(path, root) for root in roots)]
if out_of_scope:
return (
"Ownership contract blocks writes outside the assigned workspace. "
f"Out-of-scope target(s): {', '.join(out_of_scope[:4])}."
)
if tool_name in _COLLABORATION_TOOLS:
# TODO(role-identity): route through a central work-item role accessor.
from_role = str(getattr(task, "assigned_to", "") or metadata.get("work_item_role_id", "") or "").strip()
allowed = set(effective_contact_roles(from_role, task=task, org_engine=org_engine))
recipients: list[str] = []
if tool_name == "broadcast_issue":
recipients = _normalized_role_list(arguments.get("to_agents"))
elif tool_name == "start_meeting":
recipients = _normalized_role_list(arguments.get("participants"))
else:
recipients = _normalized_role_list([arguments.get("to_agent")])
invalid = [recipient for recipient in recipients if recipient and recipient not in allowed]
if invalid:
return (
"Ownership contract / collaboration topology blocks this contact target. "
f"Recipient(s) not allowed for the current work item: {', '.join(invalid)}."
)
return None
def render_ownership_contract(task: Task | None) -> str:
"""Render the boundary-enforcing part of the ownership contract.
Only three fields are rendered here: write scope, allowed
collaboration targets, and downstream consumers. The
``summary`` and ``expected_artifacts`` fields are intentionally
NOT rendered — ``summary`` duplicates the work-item identity's
``your_responsibility``, and ``expected_artifacts`` duplicates
the work-item identity's ``deliverables``. Both already appear in
the work-item identity block that every role receives, so repeating
them here would be redundant.
Returns "" if the task carries no contract or if none of the
boundary-enforcing fields are populated.
"""
if task is None:
return ""
ownership_contract = dict(getattr(task, "metadata", {}).get("ownership_contract", {}) or {})
if not ownership_contract:
return ""
lines: list[str] = []
write_scope = str(ownership_contract.get("write_scope", "") or "").strip()
if write_scope:
lines.append(f"Write scope: {write_scope}")
allowed_collaboration_targets = _normalized_role_list(ownership_contract.get("allowed_collaboration_targets"))
if allowed_collaboration_targets:
lines.append("Allowed collaboration targets:")
lines.extend(f"- {item}" for item in allowed_collaboration_targets[:12])
downstream = _normalized_role_list(ownership_contract.get("downstream_consumer"))
if downstream:
lines.append("Downstream consumers:")
lines.extend(f"- {item}" for item in downstream[:8])
if not lines:
return ""
return "## Ownership Contract\n" + "\n".join(lines)