Skip to content

feat(tasks): replace turn_complete thread writeback with pr_created artifact messages #72857

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
k11kirky merged 5 commits into from
Jul 23, 2026
Merged

feat(tasks): replace turn_complete thread writeback with pr_created artifact messages #72857

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
fba4d13
feat(tasks): replace turn_complete thread writeback with pr_created a…
k11kirky Jul 23, 2026
eea52de
fix(tasks): harden pr_created thread announcements against races and …
k11kirky Jul 23, 2026
023efb1
fix(tasks): post pr_created for channel-less tasks and log skip reasons
k11kirky Jul 23, 2026
3049ff4
fix(tasks): address pr artifact review feedback
k11kirky Jul 23, 2026
ac44ec1
chore: update OpenAPI generated types
tests-posthog[bot] Jul 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
122 changes: 80 additions & 42 deletions products/tasks/backend/facade/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
from uuid import UUID, uuid4

from django.conf import settings
from django.db import IntegrityError, close_old_connections, transaction
from django.db import IntegrityError, transaction
from django.db.models import CharField, Count, Exists, F, Min, OuterRef, Q, QuerySet, Subquery
from django.db.models.fields.json import KeyTextTransform
from django.utils import timezone as django_timezone
Expand All @@ -49,7 +49,7 @@
is_custom_images_enabled,
read_spec_from_builder_sandbox,
)
from products.tasks.backend.mentions import format_mention_token, resolve_mentioned_user_ids
from products.tasks.backend.mentions import resolve_mentioned_user_ids
from products.tasks.backend.models import (
Channel,
ChannelFeedMessage,
Expand Down Expand Up @@ -2112,6 +2112,7 @@ def update_task_run(
if new_pr_url and new_pr_url != old_pr_url:
_post_slack_update_for_pr(run)
_send_wizard_pr_ready_email_for_pr(run)
post_pr_created_thread_update(run, new_pr_url)
# Surface the PR in the run's progress timeline the moment the agent reports it, so the install
# UI advances past "Started agent" instead of waiting on the 15-min CI follow-up loop to emit
# these. Steps coalesce by id with the workflow's own pr/ci emissions (frontend mergeProgressStep),
Expand Down Expand Up @@ -2162,6 +2163,8 @@ def set_task_run_output(
run.publish_stream_state_event()
_post_slack_update_for_pr(run)
_send_wizard_pr_ready_email_for_pr(run)
if merged.get("pr_url"):
post_pr_created_thread_update(run, merged["pr_url"])
return _task_run_detail_to_dto(run)


Expand Down Expand Up @@ -5109,6 +5112,9 @@ def list_thread_messages(
return None
messages = (
TaskThreadMessage.objects.filter(task_id=task_id, team_id=team_id)
# The thread is human-to-human plus artifact announcements; rows written
# back when the agent finished a turn (a since-removed behavior) stay out.
.exclude(event="turn_complete")
Comment thread
k11kirky marked this conversation as resolved.
.select_related("author", "forwarded_by")
.order_by("created_at", "id")
)
Expand Down Expand Up @@ -5166,7 +5172,9 @@ def list_mentions(
mentioned_user_id=user_id,
# task__in keeps the visibility rules single-sourced in _visible_task_qs.
task__in=_visible_task_qs(team_id, user_id),
)
# Legacy turn_complete rows are hidden from threads (see list_thread_messages),
# so their indexed mentions must not surface notifications pointing at them.
).exclude(message__event="turn_complete")
if since is not None:
qs = qs.filter(created_at__gt=since)
mentions = qs.select_related("message__author", "task__channel").order_by("-created_at")[:limit]
Expand Down Expand Up @@ -5248,13 +5256,6 @@ def forward_thread_message(
# updates are gated on the same flag — evaluated for the task creator.
AGENT_THREAD_UPDATES_FLAG = "project-bluebird"

# One turn-complete post per run within the window, so an SSE relay reconnect
# replaying the tail of the stream can't double-post the same end-of-turn.
_TURN_COMPLETE_COOLDOWN_SECONDS = 30

# Cap the relayed final message so one agent essay can't dwarf the thread.
_TURN_MESSAGE_MAX_CHARS = 4000


def _create_agent_thread_message(task: Task, content: str, *, event: str, payload: dict | None = None) -> None:
"""Write an agent-authored thread message and index its mentions.
Expand Down Expand Up @@ -5326,42 +5327,79 @@ def post_canvas_created_thread_update(
logger.exception("Failed to post canvas-created thread update", extra={"task_id": str(task_id)})


def post_turn_complete_thread_update(
run_id: str | UUID, task_id: str | UUID, team_id: int, *, message: str | None = None
) -> None:
"""Post the agent's final turn message into the task's thread, @-mentioning the task creator.
_GITHUB_PR_PATH_PATTERN = re.compile(r"/([^/]+)/([^/]+)/pull/(\d+)/?", re.IGNORECASE)

# Characters that could break out of a markdown [label](url) token or smuggle
# extra markdown into the rendered thread message.
_PR_URL_UNSAFE_CHARS = set(" \t\n\r()[]<>\"'`\\")

_PR_URL_MAX_LENGTH = 2048


Fires from the sandbox event relay on every end-of-turn of a channel task's
background run, so the update lands even with no client open. ``message`` is
the agent's closing prose for the turn; when the relay captured none, a plain
"Turn complete." stands in. Best-effort and never raises — a failed post must
not disturb the relay.
def _is_safe_pr_url(pr_url: str) -> bool:
"""Whether ``pr_url`` is a plain http(s) URL safe to embed in a markdown link.

``pr_url`` originates from task-run output APIs, so it is caller-controlled.
Real PR URLs never contain whitespace, quotes, brackets, or parentheses;
anything that does is rejected rather than escaped.
"""
if not pr_url or len(pr_url) > _PR_URL_MAX_LENGTH or any(char in _PR_URL_UNSAFE_CHARS for char in pr_url):
return False
parsed = urlparse(pr_url)
return parsed.scheme in ("http", "https") and bool(parsed.netloc)


def _pr_display_label(pr_url: str) -> str:
parsed = urlparse(pr_url)
if parsed.hostname is None or parsed.hostname.lower() != "github.com":
return pr_url
match = _GITHUB_PR_PATH_PATTERN.fullmatch(parsed.path)
if match:
owner, repo, number = match.groups()
return f"{owner}/{repo}#{number}"
return pr_url


def post_pr_created_thread_update(run: TaskRun, pr_url: str) -> None:
"""Announce a run's freshly opened pull request in its task's thread.

Posts "[owner/repo#N](url) has been opened" as an agent artifact message
(``event="pr_created"``). Both the agent-output path and the GitHub webhook
backstop can observe the same PR, so the announcement dedupes on the task's
existing ``pr_created`` rows for this URL. Best-effort and never raises —
recording the PR must not fail because its announcement couldn't be written.
"""
try:
if not settings.TEST:
close_old_connections()
task = Task.objects.select_related("created_by").filter(id=task_id, team_id=team_id).first()
# Threads hang off a task's channel feed; a channel-less task has no audience.
if task is None or task.channel_id is None:
if not _is_safe_pr_url(pr_url):
logger.info("pr_created thread update skipped", extra={"task_id": str(run.task_id), "reason": "unsafe_url"})
return
creator = task.created_by
if creator is None or not _agent_thread_updates_enabled(creator):
# Unlike turn_complete's old channel guard, artifact rows post for
# channel-less tasks too: every task has a thread panel.
task = Task.objects.select_related("created_by").filter(id=run.task_id, team_id=run.team_id).first()
if task is None:
return
from products.tasks.backend.redis import get_tasks_cache # noqa: PLC0415 — keep redis off the api import path

if not get_tasks_cache().add(f"thread_update:{run_id}:turn_complete", True, _TURN_COMPLETE_COOLDOWN_SECONDS):
if task.created_by is None or not _agent_thread_updates_enabled(task.created_by):
logger.info(
"pr_created thread update skipped",
extra={"task_id": str(task.id), "reason": "no_creator" if task.created_by is None else "flag_off"},
)
return
body = (message or "").strip() or "Turn complete."
if len(body) > _TURN_MESSAGE_MAX_CHARS:
body = body[: _TURN_MESSAGE_MAX_CHARS - 1] + "…"
mention = format_mention_token(creator.get_full_name() or creator.email, creator.email)
# payload.run_id is the dedupe key: a client already rendering this run's
# live agent turns can suppress the durable row (or vice versa).
_create_agent_thread_message(
task,
f"{mention} {body}",
event="turn_complete",
payload={"run_id": str(run_id)},
)
# The agent-output path and the webhook backstop can race on the same PR;
# locking the task row makes the dedupe check-and-create atomic across them.
with transaction.atomic():
Task.objects.select_for_update().filter(id=task.id).first()
if (
TaskThreadMessage.objects.for_team(task.team_id)
.filter(task_id=task.id, event="pr_created", payload__pr_url=pr_url)
.exists()
):
return
label = _pr_display_label(pr_url)
_create_agent_thread_message(
task,
f"[{label}]({pr_url}) has been opened",
event="pr_created",
payload={"pr_url": pr_url},
)
except Exception:
logger.exception("Failed to post turn-complete thread update", extra={"task_id": str(task_id)})
logger.exception("Failed to post pr-created thread update", extra={"task_id": str(run.task_id)})
41 changes: 0 additions & 41 deletions products/tasks/backend/temporal/process_task/activities/relay_sandbox_events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@

from posthog.temporal.common.utils import close_db_connections

from products.tasks.backend.facade import api as tasks_facade
from products.tasks.backend.logic.services.agent_command import validate_sandbox_url
from products.tasks.backend.logic.services.connection_token import create_sandbox_connection_token
from products.tasks.backend.logic.services.permission_broker import (
Expand Down Expand Up @@ -331,10 +330,6 @@ async def _relay_loop(
last_audit_ts_ns: list[int] = [0] # track last agentsh audit timestamp
# Brackets turn_started / turn_completed signals to the parent.
slack_turn_active: list[bool] = [False]
# The agent's in-progress closing message for the current turn. Chunks
# accumulate; a new tool call or user message resets it, so at end-of-turn
# it holds the prose after the last tool call — what the thread update posts.
final_message_parts: list[str] = []
# ACP emits one tool_call + N tool_call_update per id; only render the start.
emitted_tool_call_ids: set[str] = set()
# Buffered prose + last flush time (monotonic); see TEXT_DELTA_FLUSH_INTERVAL_SECONDS.
Expand Down Expand Up @@ -409,22 +404,6 @@ async def _relay_loop(
# does sync Redis (cache.add) and a potential network call to
# the feature-flag service.
asyncio.create_task(asyncio.to_thread(_safe_dispatch_awaiting_input, task_run))
if task_run is not None and task_run.mode != "interactive":
# Background run finished a turn — post its closing message
# into the task's thread so teammates following it see the
# outcome without a client open. Guards (flag, channel,
# cooldown) live in the facade; same thread hop as above
# for its sync I/O.
asyncio.create_task(
asyncio.to_thread(
tasks_facade.post_turn_complete_thread_update,
str(task_run.id),
str(task_run.task_id),
task_run.team_id,
message="".join(final_message_parts).strip() or None,
)
)
final_message_parts.clear()
if is_agent_design_enabled and slack_turn_active[0] and workflow_handle is not None:
slack_turn_active[0] = False
# Awaited in order: the final prose must be recorded before
Expand All @@ -435,9 +414,6 @@ async def _relay_loop(
elif not agent_active[0] and _is_active_agent_update(event_data):
agent_active[0] = True

if task_run is not None and task_run.mode != "interactive":
_track_final_message(event_data, final_message_parts)

# Agent-design signal fan-out: first session/update opens the
# child relay; tool_call → step, agent_message_chunk → markdown.
if is_agent_design_enabled and workflow_handle is not None:
Expand Down Expand Up @@ -500,7 +476,6 @@ async def _relay_loop(
reconnect_count += 1
# May have missed an end_of_turn on the dropped stream — assume idle until re-confirmed.
agent_active[0] = False
final_message_parts.clear()
# Drop un-flushed partial prose — the agent replays events on reconnect.
pending_text_parts.clear()
logger.warning(
Expand All @@ -524,7 +499,6 @@ async def _relay_loop(
# 5xx — transient server error, worth retrying
reconnect_count += 1
agent_active[0] = False # missed-end_of_turn guard (see ReadTimeout above)
final_message_parts.clear()
# Drop un-flushed partial prose — the agent replays events on reconnect.
pending_text_parts.clear()
logger.warning(
Expand All @@ -539,7 +513,6 @@ async def _relay_loop(
except (httpx.TransportError, httpx_sse.SSEError) as e:
reconnect_count += 1
agent_active[0] = False # missed-end_of_turn guard (see ReadTimeout above)
final_message_parts.clear()
# Drop un-flushed partial prose — the agent replays events on reconnect.
pending_text_parts.clear()
logger.warning(
Expand Down Expand Up @@ -671,20 +644,6 @@ def _tool_args_preview(raw_input: Any) -> str | None:
return one_line


def _track_final_message(event_data: dict, parts: list[str]) -> None:
"""Accumulate agent_message_chunk text; a new tool call or user message resets,
so `parts` ends the turn holding only the agent's closing prose."""
text = _extract_agent_message_text(event_data)
if text:
parts.append(text)
return
if not _is_session_update(event_data):
return
update = (event_data.get("notification", {}).get("params") or {}).get("update") or {}
if update.get("sessionUpdate") in ("tool_call", "user_message", "user_message_chunk"):
parts.clear()


def _extract_agent_message_text(event_data: dict) -> str | None:
"""Text delta from an ACP agent_message_chunk session/update, else None."""
notification = event_data.get("notification", {})
Expand Down
Loading
Loading