630 lines
23 KiB
Python
630 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
|
|
from asgiref.sync import sync_to_async
|
|
from django.utils import timezone
|
|
|
|
from core.commands.base import CommandContext, CommandHandler, CommandResult
|
|
from core.commands.delivery import post_status_in_source
|
|
from core.messaging.text_export import plain_text_blob
|
|
from core.models import (
|
|
ChatTaskSource,
|
|
CodexPermissionRequest,
|
|
CodexRun,
|
|
CommandProfile,
|
|
DerivedTask,
|
|
ExternalSyncEvent,
|
|
Message,
|
|
TaskProject,
|
|
TaskProviderConfig,
|
|
)
|
|
from core.tasks.codex_approval import queue_codex_event_with_pre_approval
|
|
from core.tasks.codex_support import channel_variants, resolve_external_chat_id
|
|
|
|
_CLAUDE_DEFAULT_RE = re.compile(
|
|
r"^\s*(?:\.claude\b|#claude#?)(?P<body>.*)$",
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
_CLAUDE_PLAN_RE = re.compile(
|
|
r"^\s*(?:\.claude\s+plan\b|#claude\s+plan#?)(?P<body>.*)$",
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
_CLAUDE_STATUS_RE = re.compile(
|
|
r"^\s*(?:\.claude\s+status\b|#claude\s+status#?)\s*$", re.IGNORECASE
|
|
)
|
|
_CLAUDE_APPROVE_DENY_RE = re.compile(
|
|
r"^\s*(?:\.claude|#claude)\s+(?P<action>approve|deny)\s+(?P<approval_key>[A-Za-z0-9._:-]+)#?\s*$",
|
|
re.IGNORECASE,
|
|
)
|
|
_PROJECT_TOKEN_RE = re.compile(r"\[\s*project\s*:\s*([^\]]+)\]", re.IGNORECASE)
|
|
_REFERENCE_RE = re.compile(r"(?<!\w)#([A-Za-z0-9_-]+)\b")
|
|
|
|
|
|
class ClaudeParsedCommand(dict):
|
|
@property
|
|
def command(self) -> str | None:
|
|
value = self.get("command")
|
|
return str(value) if value else None
|
|
|
|
@property
|
|
def body_text(self) -> str:
|
|
return str(self.get("body_text") or "")
|
|
|
|
@property
|
|
def approval_key(self) -> str:
|
|
return str(self.get("approval_key") or "")
|
|
|
|
|
|
def parse_claude_command(text: str) -> ClaudeParsedCommand:
|
|
body = str(text or "")
|
|
m = _CLAUDE_APPROVE_DENY_RE.match(body)
|
|
if m:
|
|
return ClaudeParsedCommand(
|
|
command=str(m.group("action") or "").strip().lower(),
|
|
body_text="",
|
|
approval_key=str(m.group("approval_key") or "").strip(),
|
|
)
|
|
if _CLAUDE_STATUS_RE.match(body):
|
|
return ClaudeParsedCommand(command="status", body_text="", approval_key="")
|
|
m = _CLAUDE_PLAN_RE.match(body)
|
|
if m:
|
|
return ClaudeParsedCommand(
|
|
command="plan",
|
|
body_text=str(m.group("body") or "").strip(),
|
|
approval_key="",
|
|
)
|
|
m = _CLAUDE_DEFAULT_RE.match(body)
|
|
if m:
|
|
return ClaudeParsedCommand(
|
|
command="default",
|
|
body_text=str(m.group("body") or "").strip(),
|
|
approval_key="",
|
|
)
|
|
return ClaudeParsedCommand(command=None, body_text="", approval_key="")
|
|
|
|
|
|
def claude_trigger_matches(
|
|
message_text: str, trigger_token: str, exact_match_only: bool
|
|
) -> bool:
|
|
body = str(message_text or "").strip()
|
|
parsed = parse_claude_command(body)
|
|
if parsed.command:
|
|
return True
|
|
trigger = str(trigger_token or "").strip()
|
|
if not trigger:
|
|
return False
|
|
if exact_match_only:
|
|
return body.lower() == trigger.lower()
|
|
return trigger.lower() in body.lower()
|
|
|
|
|
|
class ClaudeCommandHandler(CommandHandler):
|
|
slug = "claude"
|
|
_provider_name = "claude_cli"
|
|
_approval_prefix = "claude_approval"
|
|
|
|
async def _load_trigger(self, message_id: str) -> Message | None:
|
|
return await sync_to_async(
|
|
lambda: Message.objects.select_related(
|
|
"user", "session", "session__identifier", "reply_to"
|
|
)
|
|
.filter(id=message_id)
|
|
.first()
|
|
)()
|
|
|
|
def _effective_scope(self, trigger: Message) -> tuple[str, str]:
|
|
service = str(getattr(trigger, "source_service", "") or "").strip().lower()
|
|
channel = str(getattr(trigger, "source_chat_id", "") or "").strip()
|
|
identifier = getattr(getattr(trigger, "session", None), "identifier", None)
|
|
fallback_service = str(getattr(identifier, "service", "") or "").strip().lower()
|
|
fallback_identifier = str(getattr(identifier, "identifier", "") or "").strip()
|
|
if (
|
|
service == "web"
|
|
and fallback_service
|
|
and fallback_identifier
|
|
and fallback_service != "web"
|
|
):
|
|
return fallback_service, fallback_identifier
|
|
return service or "web", channel
|
|
|
|
async def _mapped_sources(
|
|
self, user, service: str, channel: str
|
|
) -> list[ChatTaskSource]:
|
|
variants = channel_variants(service, channel)
|
|
if not variants:
|
|
return []
|
|
return await sync_to_async(list)(
|
|
ChatTaskSource.objects.filter(
|
|
user=user,
|
|
enabled=True,
|
|
service=service,
|
|
channel_identifier__in=variants,
|
|
).select_related("project", "epic")
|
|
)
|
|
|
|
async def _linked_task_from_reply(
|
|
self, user, reply_to: Message | None
|
|
) -> DerivedTask | None:
|
|
if reply_to is None:
|
|
return None
|
|
by_origin = await sync_to_async(
|
|
lambda: DerivedTask.objects.filter(user=user, origin_message=reply_to)
|
|
.select_related("project", "epic")
|
|
.order_by("-created_at")
|
|
.first()
|
|
)()
|
|
if by_origin is not None:
|
|
return by_origin
|
|
return await sync_to_async(
|
|
lambda: DerivedTask.objects.filter(
|
|
user=user, events__source_message=reply_to
|
|
)
|
|
.select_related("project", "epic")
|
|
.order_by("-created_at")
|
|
.first()
|
|
)()
|
|
|
|
def _extract_project_token(self, body_text: str) -> tuple[str, str]:
|
|
text = str(body_text or "")
|
|
m = _PROJECT_TOKEN_RE.search(text)
|
|
if not m:
|
|
return "", text
|
|
token = str(m.group(1) or "").strip()
|
|
cleaned = _PROJECT_TOKEN_RE.sub("", text).strip()
|
|
return token, cleaned
|
|
|
|
def _extract_reference(self, body_text: str) -> str:
|
|
m = _REFERENCE_RE.search(str(body_text or ""))
|
|
if not m:
|
|
return ""
|
|
return str(m.group(1) or "").strip()
|
|
|
|
async def _resolve_task(
|
|
self, user, reference_code: str, reply_task: DerivedTask | None
|
|
) -> DerivedTask | None:
|
|
if reference_code:
|
|
return await sync_to_async(
|
|
lambda: DerivedTask.objects.filter(
|
|
user=user, reference_code=reference_code
|
|
)
|
|
.select_related("project", "epic")
|
|
.order_by("-created_at")
|
|
.first()
|
|
)()
|
|
return reply_task
|
|
|
|
async def _resolve_project(
|
|
self,
|
|
*,
|
|
user,
|
|
service: str,
|
|
channel: str,
|
|
task: DerivedTask | None,
|
|
reply_task: DerivedTask | None,
|
|
project_token: str,
|
|
) -> tuple[TaskProject | None, str]:
|
|
if task is not None:
|
|
return task.project, ""
|
|
if reply_task is not None:
|
|
return reply_task.project, ""
|
|
if project_token:
|
|
project = await sync_to_async(
|
|
lambda: TaskProject.objects.filter(
|
|
user=user, name__iexact=project_token
|
|
).first()
|
|
)()
|
|
if project is not None:
|
|
return project, ""
|
|
return None, f"project_not_found:{project_token}"
|
|
|
|
mapped = await self._mapped_sources(user, service, channel)
|
|
project_ids = sorted({str(row.project_id) for row in mapped if row.project_id})
|
|
if len(project_ids) == 1:
|
|
project = next(
|
|
(
|
|
row.project
|
|
for row in mapped
|
|
if str(row.project_id) == project_ids[0]
|
|
),
|
|
None,
|
|
)
|
|
return project, ""
|
|
if len(project_ids) > 1:
|
|
return None, "project_required:[project:Name]"
|
|
return None, "project_unresolved"
|
|
|
|
async def _post_source_status(
|
|
self, trigger: Message, text: str, suffix: str
|
|
) -> None:
|
|
await post_status_in_source(
|
|
trigger_message=trigger,
|
|
text=text,
|
|
origin_tag=f"claude-status:{suffix}",
|
|
)
|
|
|
|
async def _run_status(
|
|
self, trigger: Message, service: str, channel: str, project: TaskProject | None
|
|
) -> CommandResult:
|
|
def _load_runs():
|
|
qs = CodexRun.objects.filter(user=trigger.user)
|
|
if service:
|
|
qs = qs.filter(source_service=service)
|
|
if channel:
|
|
qs = qs.filter(source_channel=channel)
|
|
if project is not None:
|
|
qs = qs.filter(project=project)
|
|
return list(qs.order_by("-created_at")[:10])
|
|
|
|
runs = await sync_to_async(_load_runs)()
|
|
if not runs:
|
|
await self._post_source_status(
|
|
trigger, "[claude] no recent runs for this scope.", "empty"
|
|
)
|
|
return CommandResult(ok=True, status="ok", payload={"count": 0})
|
|
lines = ["[claude] recent runs:"]
|
|
for row in runs:
|
|
ref = str(getattr(getattr(row, "task", None), "reference_code", "") or "-")
|
|
summary = str((row.result_payload or {}).get("summary") or "").strip()
|
|
summary_part = f" · {summary}" if summary else ""
|
|
lines.append(f"- {row.status} run={row.id} task=#{ref}{summary_part}")
|
|
await self._post_source_status(trigger, "\n".join(lines), "runs")
|
|
return CommandResult(ok=True, status="ok", payload={"count": len(runs)})
|
|
|
|
async def _run_approval_action(
|
|
self,
|
|
trigger: Message,
|
|
parsed: ClaudeParsedCommand,
|
|
current_service: str,
|
|
current_channel: str,
|
|
) -> CommandResult:
|
|
cfg = await sync_to_async(
|
|
lambda: TaskProviderConfig.objects.filter(
|
|
user=trigger.user, provider=self._provider_name
|
|
).first()
|
|
)()
|
|
settings_payload = dict(getattr(cfg, "settings", {}) or {})
|
|
approver_service = (
|
|
str(settings_payload.get("approver_service") or "").strip().lower()
|
|
)
|
|
approver_identifier = str(
|
|
settings_payload.get("approver_identifier") or ""
|
|
).strip()
|
|
if not approver_service or not approver_identifier:
|
|
return CommandResult(
|
|
ok=False, status="failed", error="approver_channel_not_configured"
|
|
)
|
|
|
|
if str(current_service or "").strip().lower() != approver_service or str(
|
|
current_channel or ""
|
|
).strip() not in set(channel_variants(approver_service, approver_identifier)):
|
|
return CommandResult(
|
|
ok=False,
|
|
status="failed",
|
|
error="approval_command_not_allowed_in_this_channel",
|
|
)
|
|
|
|
approval_key = parsed.approval_key
|
|
request = await sync_to_async(
|
|
lambda: CodexPermissionRequest.objects.select_related(
|
|
"codex_run", "external_sync_event"
|
|
)
|
|
.filter(user=trigger.user, approval_key=approval_key)
|
|
.first()
|
|
)()
|
|
if request is None:
|
|
return CommandResult(
|
|
ok=False, status="failed", error="approval_key_not_found"
|
|
)
|
|
|
|
now = timezone.now()
|
|
if parsed.command == "approve":
|
|
request.status = "approved"
|
|
request.resolved_at = now
|
|
request.resolved_by_identifier = current_channel
|
|
request.resolution_note = "approved via claude command"
|
|
await sync_to_async(request.save)(
|
|
update_fields=[
|
|
"status",
|
|
"resolved_at",
|
|
"resolved_by_identifier",
|
|
"resolution_note",
|
|
]
|
|
)
|
|
if request.external_sync_event_id:
|
|
await sync_to_async(
|
|
ExternalSyncEvent.objects.filter(
|
|
id=request.external_sync_event_id
|
|
).update
|
|
)(
|
|
status="ok",
|
|
error="",
|
|
)
|
|
run = request.codex_run
|
|
run.status = "approved_waiting_resume"
|
|
run.error = ""
|
|
await sync_to_async(run.save)(
|
|
update_fields=["status", "error", "updated_at"]
|
|
)
|
|
source_service = str(run.source_service or "")
|
|
source_channel = str(run.source_channel or "")
|
|
resume_payload = dict(request.resume_payload or {})
|
|
resume_action = str(resume_payload.get("action") or "").strip().lower()
|
|
resume_provider_payload = dict(resume_payload.get("provider_payload") or {})
|
|
if resume_action and resume_provider_payload:
|
|
provider_payload = dict(resume_provider_payload)
|
|
provider_payload["codex_run_id"] = str(run.id)
|
|
provider_payload["source_service"] = source_service
|
|
provider_payload["source_channel"] = source_channel
|
|
event_action = resume_action
|
|
resume_idempotency_key = str(
|
|
resume_payload.get("idempotency_key") or ""
|
|
).strip()
|
|
resume_event_key = (
|
|
resume_idempotency_key
|
|
if resume_idempotency_key
|
|
else f"{self._approval_prefix}:{approval_key}:approved"
|
|
)
|
|
else:
|
|
provider_payload = dict(
|
|
run.request_payload.get("provider_payload") or {}
|
|
)
|
|
provider_payload.update(
|
|
{
|
|
"mode": "approval_response",
|
|
"approval_key": approval_key,
|
|
"resume_payload": dict(request.resume_payload or {}),
|
|
"codex_run_id": str(run.id),
|
|
"source_service": source_service,
|
|
"source_channel": source_channel,
|
|
}
|
|
)
|
|
event_action = "append_update"
|
|
resume_event_key = f"{self._approval_prefix}:{approval_key}:approved"
|
|
await sync_to_async(ExternalSyncEvent.objects.update_or_create)(
|
|
idempotency_key=resume_event_key,
|
|
defaults={
|
|
"user": trigger.user,
|
|
"task_id": run.task_id,
|
|
"task_event_id": run.derived_task_event_id,
|
|
"provider": self._provider_name,
|
|
"status": "pending",
|
|
"payload": {
|
|
"action": event_action,
|
|
"provider_payload": provider_payload,
|
|
},
|
|
"error": "",
|
|
},
|
|
)
|
|
return CommandResult(
|
|
ok=True,
|
|
status="ok",
|
|
payload={"approval_key": approval_key, "resolution": "approved"},
|
|
)
|
|
|
|
request.status = "denied"
|
|
request.resolved_at = now
|
|
request.resolved_by_identifier = current_channel
|
|
request.resolution_note = "denied via claude command"
|
|
await sync_to_async(request.save)(
|
|
update_fields=[
|
|
"status",
|
|
"resolved_at",
|
|
"resolved_by_identifier",
|
|
"resolution_note",
|
|
]
|
|
)
|
|
if request.external_sync_event_id:
|
|
await sync_to_async(
|
|
ExternalSyncEvent.objects.filter(
|
|
id=request.external_sync_event_id
|
|
).update
|
|
)(
|
|
status="failed",
|
|
error="approval_denied",
|
|
)
|
|
run = request.codex_run
|
|
run.status = "denied"
|
|
run.error = "approval_denied"
|
|
await sync_to_async(run.save)(update_fields=["status", "error", "updated_at"])
|
|
await sync_to_async(ExternalSyncEvent.objects.update_or_create)(
|
|
idempotency_key=f"{self._approval_prefix}:{approval_key}:denied",
|
|
defaults={
|
|
"user": trigger.user,
|
|
"task_id": run.task_id,
|
|
"task_event_id": run.derived_task_event_id,
|
|
"provider": self._provider_name,
|
|
"status": "failed",
|
|
"payload": {
|
|
"action": "append_update",
|
|
"provider_payload": {
|
|
"mode": "approval_response",
|
|
"approval_key": approval_key,
|
|
"codex_run_id": str(run.id),
|
|
},
|
|
},
|
|
"error": "approval_denied",
|
|
},
|
|
)
|
|
return CommandResult(
|
|
ok=True,
|
|
status="ok",
|
|
payload={"approval_key": approval_key, "resolution": "denied"},
|
|
)
|
|
|
|
async def _create_submission(
|
|
self,
|
|
*,
|
|
trigger: Message,
|
|
mode: str,
|
|
body_text: str,
|
|
task: DerivedTask,
|
|
project: TaskProject,
|
|
) -> CommandResult:
|
|
cfg = await sync_to_async(
|
|
lambda: TaskProviderConfig.objects.filter(
|
|
user=trigger.user, provider=self._provider_name, enabled=True
|
|
).first()
|
|
)()
|
|
if cfg is None:
|
|
return CommandResult(
|
|
ok=False, status="failed", error="provider_disabled_or_missing"
|
|
)
|
|
|
|
service, channel = self._effective_scope(trigger)
|
|
external_chat_id = await sync_to_async(resolve_external_chat_id)(
|
|
user=trigger.user,
|
|
provider=self._provider_name,
|
|
service=service,
|
|
channel=channel,
|
|
)
|
|
payload = {
|
|
"task_id": str(task.id),
|
|
"reference_code": str(task.reference_code or ""),
|
|
"title": str(task.title or ""),
|
|
"external_key": str(task.external_key or ""),
|
|
"project_name": str(getattr(project, "name", "") or ""),
|
|
"epic_name": str(getattr(getattr(task, "epic", None), "name", "") or ""),
|
|
"source_service": service,
|
|
"source_channel": channel,
|
|
"external_chat_id": external_chat_id,
|
|
"origin_message_id": str(getattr(task, "origin_message_id", "") or ""),
|
|
"trigger_message_id": str(trigger.id),
|
|
"mode": mode,
|
|
"command_text": str(body_text or ""),
|
|
}
|
|
if mode == "plan":
|
|
anchor = trigger.reply_to
|
|
if anchor is None:
|
|
return CommandResult(
|
|
ok=False, status="failed", error="reply_required_for_claude_plan"
|
|
)
|
|
rows = await sync_to_async(list)(
|
|
Message.objects.filter(
|
|
user=trigger.user,
|
|
session=trigger.session,
|
|
ts__gte=int(anchor.ts or 0),
|
|
ts__lte=int(trigger.ts or 0),
|
|
)
|
|
.order_by("ts")
|
|
.select_related(
|
|
"session", "session__identifier", "session__identifier__person"
|
|
)
|
|
)
|
|
payload["reply_context"] = {
|
|
"anchor_message_id": str(anchor.id),
|
|
"trigger_message_id": str(trigger.id),
|
|
"message_ids": [str(row.id) for row in rows],
|
|
"content": plain_text_blob(rows),
|
|
}
|
|
|
|
run = await sync_to_async(CodexRun.objects.create)(
|
|
user=trigger.user,
|
|
task=task,
|
|
source_message=trigger,
|
|
project=project,
|
|
epic=getattr(task, "epic", None),
|
|
source_service=service,
|
|
source_channel=channel,
|
|
external_chat_id=external_chat_id,
|
|
status="waiting_approval",
|
|
request_payload={
|
|
"action": "append_update",
|
|
"provider_payload": dict(payload),
|
|
},
|
|
result_payload={},
|
|
error="",
|
|
)
|
|
payload["codex_run_id"] = str(run.id)
|
|
run.request_payload = {
|
|
"action": "append_update",
|
|
"provider_payload": dict(payload),
|
|
}
|
|
await sync_to_async(run.save)(update_fields=["request_payload", "updated_at"])
|
|
|
|
idempotency_key = f"claude_cmd:{trigger.id}:{mode}:{task.id}:{hashlib.sha1(str(body_text or '').encode('utf-8')).hexdigest()[:12]}"
|
|
await sync_to_async(queue_codex_event_with_pre_approval)(
|
|
user=trigger.user,
|
|
run=run,
|
|
task=task,
|
|
task_event=None,
|
|
action="append_update",
|
|
provider_payload=dict(payload),
|
|
idempotency_key=idempotency_key,
|
|
)
|
|
return CommandResult(
|
|
ok=True,
|
|
status="ok",
|
|
payload={"codex_run_id": str(run.id), "approval_required": True},
|
|
)
|
|
|
|
async def execute(self, ctx: CommandContext) -> CommandResult:
|
|
trigger = await self._load_trigger(ctx.message_id)
|
|
if trigger is None:
|
|
return CommandResult(ok=False, status="failed", error="trigger_not_found")
|
|
|
|
profile = await sync_to_async(
|
|
lambda: CommandProfile.objects.filter(
|
|
user=trigger.user, slug=self.slug, enabled=True
|
|
).first()
|
|
)()
|
|
if profile is None:
|
|
return CommandResult(ok=False, status="skipped", error="profile_missing")
|
|
|
|
parsed = parse_claude_command(ctx.message_text)
|
|
if not parsed.command:
|
|
return CommandResult(
|
|
ok=False, status="skipped", error="claude_command_not_matched"
|
|
)
|
|
|
|
service, channel = self._effective_scope(trigger)
|
|
|
|
if parsed.command == "status":
|
|
project = None
|
|
reply_task = await self._linked_task_from_reply(
|
|
trigger.user, trigger.reply_to
|
|
)
|
|
if reply_task is not None:
|
|
project = reply_task.project
|
|
return await self._run_status(trigger, service, channel, project)
|
|
|
|
if parsed.command in {"approve", "deny"}:
|
|
return await self._run_approval_action(
|
|
trigger,
|
|
parsed,
|
|
current_service=str(ctx.service or ""),
|
|
current_channel=str(ctx.channel_identifier or ""),
|
|
)
|
|
|
|
project_token, cleaned_body = self._extract_project_token(parsed.body_text)
|
|
reference_code = self._extract_reference(cleaned_body)
|
|
reply_task = await self._linked_task_from_reply(trigger.user, trigger.reply_to)
|
|
task = await self._resolve_task(trigger.user, reference_code, reply_task)
|
|
if task is None:
|
|
return CommandResult(
|
|
ok=False, status="failed", error="task_target_required"
|
|
)
|
|
|
|
project, project_error = await self._resolve_project(
|
|
user=trigger.user,
|
|
service=service,
|
|
channel=channel,
|
|
task=task,
|
|
reply_task=reply_task,
|
|
project_token=project_token,
|
|
)
|
|
if project is None:
|
|
return CommandResult(
|
|
ok=False, status="failed", error=project_error or "project_unresolved"
|
|
)
|
|
|
|
mode = "plan" if parsed.command == "plan" else "default"
|
|
return await self._create_submission(
|
|
trigger=trigger,
|
|
mode=mode,
|
|
body_text=cleaned_body,
|
|
task=task,
|
|
project=project,
|
|
)
|