From b9ee1be885f480fa7803a83880cb268cfebd509a Mon Sep 17 00:00:00 2001 From: Hampton Lintorn-Catlin Date: Fri, 7 Aug 2026 16:56:57 -0500 Subject: [PATCH 01/35] Agent collaboration: live event inbox, presence pill, diff flashes, bridge Agents that author plans can now hear about comments the moment they land and collaborate visibly: - AgentEvent inbox + AgentSession (Linear-style pending/active/ awaiting_input/complete/stale) with fan-out from the existing notification choke point; agents never woken by their own activity. - GET /api/v1/agent/events long-poll + SSE with UUIDv7 cursor resume and explicit ack; POST/PATCH/DELETE /api/v1/plans/:id/agent_session drives a live presence pill on the plan masthead. - Content broadcasts now carry changed-section keys; live_update flashes changed blocks with word-level ins/del diffs that settle after ~2.5s. - API ergonomics: GET single comment thread, dismiss route alias (docs said dismiss, router said discard), API threads get the same initial- status rule as the web flow, agent_name on ApiToken. - script/coplan-bridge: harness-agnostic daemon (claude/codex/goose/ openhands/amp adapters + in-process demo agent) that drains the inbox and resumes your local harness session per event. - /agent-instructions documents the realtime loop and session etiquette. Co-Authored-By: Claude Fable 5 --- ...eate_agent_collaboration_tables.co_plan.rb | 46 ++++ db/schema.rb | 34 ++- .../assets/stylesheets/coplan/application.css | 99 +++++++++ .../coplan/api/v1/agent_events_controller.rb | 117 ++++++++++ .../api/v1/agent_sessions_controller.rb | 79 +++++++ .../coplan/api/v1/comments_controller.rb | 48 ++++- .../coplan/live_update_controller.js | 176 +++++++++++++++- .../coplan/mark_stale_agent_session_job.rb | 18 ++ engine/app/models/coplan/agent_event.rb | 36 ++++ engine/app/models/coplan/agent_session.rb | 63 ++++++ .../services/coplan/agent_events/publish.rb | 58 +++++ engine/app/services/coplan/broadcaster.rb | 10 +- .../services/coplan/notifications/create.rb | 26 +++ .../services/coplan/plans/replace_content.rb | 17 +- .../coplan/agent_instructions/show.text.erb | 80 ++++++- .../coplan/plans/_agent_sessions.html.erb | 13 ++ engine/app/views/coplan/plans/show.html.erb | 1 + engine/config/routes.rb | 16 +- ...00000_create_agent_collaboration_tables.rb | 45 ++++ script/coplan-bridge | 199 ++++++++++++++++++ spec/requests/api/v1/agent_events_spec.rb | 165 +++++++++++++++ spec/services/plans/replace_content_spec.rb | 3 +- 22 files changed, 1338 insertions(+), 11 deletions(-) create mode 100644 db/migrate/20260807215107_create_agent_collaboration_tables.co_plan.rb create mode 100644 engine/app/controllers/coplan/api/v1/agent_events_controller.rb create mode 100644 engine/app/controllers/coplan/api/v1/agent_sessions_controller.rb create mode 100644 engine/app/jobs/coplan/mark_stale_agent_session_job.rb create mode 100644 engine/app/models/coplan/agent_event.rb create mode 100644 engine/app/models/coplan/agent_session.rb create mode 100644 engine/app/services/coplan/agent_events/publish.rb create mode 100644 engine/app/views/coplan/plans/_agent_sessions.html.erb create mode 100644 engine/db/migrate/20260807000000_create_agent_collaboration_tables.rb create mode 100644 script/coplan-bridge create mode 100644 spec/requests/api/v1/agent_events_spec.rb diff --git a/db/migrate/20260807215107_create_agent_collaboration_tables.co_plan.rb b/db/migrate/20260807215107_create_agent_collaboration_tables.co_plan.rb new file mode 100644 index 00000000..e4377e0a --- /dev/null +++ b/db/migrate/20260807215107_create_agent_collaboration_tables.co_plan.rb @@ -0,0 +1,46 @@ +# This migration comes from co_plan (originally 20260807000000) +class CreateAgentCollaborationTables < ActiveRecord::Migration[8.1] + def change + # Durable per-agent event inbox. IDs are UUIDv7 (time-ordered), so the + # id doubles as the pagination cursor: "give me events after ". + create_table :coplan_agent_events, id: { type: :string, limit: 36 } do |t| + t.string :api_token_id, limit: 36, null: false + t.string :plan_id, limit: 36, null: false + t.string :comment_thread_id, limit: 36 + t.string :comment_id, limit: 36 + t.string :event_type, null: false + t.json :payload + t.datetime :acked_at + t.datetime :created_at, null: false + + t.index [:api_token_id, :id] + t.index [:api_token_id, :acked_at] + t.index :plan_id + end + + # One session per (plan, agent token) — Linear-style delegation state + # machine driving the presence pill: pending / active / awaiting_input / + # complete / stale. + create_table :coplan_agent_sessions, id: { type: :string, limit: 36 } do |t| + t.string :plan_id, limit: 36, null: false + t.string :api_token_id, limit: 36, null: false + t.string :agent_name, null: false + t.string :state, null: false, default: "pending" + t.string :state_detail + t.datetime :last_activity_at + t.timestamps + + t.index [:plan_id, :api_token_id], unique: true + t.index :api_token_id + end + + # Stable display identity for an agent token, instead of the free-text + # per-comment agent_name. + add_column :coplan_api_tokens, :agent_name, :string + + add_foreign_key :coplan_agent_events, :coplan_api_tokens, column: :api_token_id + add_foreign_key :coplan_agent_events, :coplan_plans, column: :plan_id + add_foreign_key :coplan_agent_sessions, :coplan_api_tokens, column: :api_token_id + add_foreign_key :coplan_agent_sessions, :coplan_plans, column: :plan_id + end +end diff --git a/db/schema.rb b/db/schema.rb index e83c344e..0ee385cd 100644 --- a/db/schema.rb +++ b/db/schema.rb @@ -10,7 +10,7 @@ # # It's strongly recommended that you check this file into your version control system. -ActiveRecord::Schema[8.1].define(version: 2026_07_19_165429) do +ActiveRecord::Schema[8.1].define(version: 2026_08_07_215107) do create_table "active_admin_comments", id: { type: :string, limit: 36 }, charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t| t.bigint "author_id" t.string "author_type" @@ -53,7 +53,35 @@ t.index ["blob_id", "variation_digest"], name: "index_active_storage_variant_records_uniqueness", unique: true end + create_table "coplan_agent_events", id: { type: :string, limit: 36 }, charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t| + t.datetime "acked_at" + t.string "api_token_id", limit: 36, null: false + t.string "comment_id", limit: 36 + t.string "comment_thread_id", limit: 36 + t.datetime "created_at", null: false + t.string "event_type", null: false + t.json "payload" + t.string "plan_id", limit: 36, null: false + t.index ["api_token_id", "acked_at"], name: "index_coplan_agent_events_on_api_token_id_and_acked_at" + t.index ["api_token_id", "id"], name: "index_coplan_agent_events_on_api_token_id_and_id" + t.index ["plan_id"], name: "index_coplan_agent_events_on_plan_id" + end + + create_table "coplan_agent_sessions", id: { type: :string, limit: 36 }, charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t| + t.string "agent_name", null: false + t.string "api_token_id", limit: 36, null: false + t.datetime "created_at", null: false + t.datetime "last_activity_at" + t.string "plan_id", limit: 36, null: false + t.string "state", default: "pending", null: false + t.string "state_detail" + t.datetime "updated_at", null: false + t.index ["api_token_id"], name: "index_coplan_agent_sessions_on_api_token_id" + t.index ["plan_id", "api_token_id"], name: "index_coplan_agent_sessions_on_plan_id_and_api_token_id", unique: true + end + create_table "coplan_api_tokens", id: { type: :string, limit: 36 }, charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t| + t.string "agent_name" t.datetime "created_at", null: false t.timestamp "expires_at" t.timestamp "last_used_at" @@ -363,6 +391,10 @@ add_foreign_key "active_storage_attachments", "active_storage_blobs", column: "blob_id" add_foreign_key "active_storage_variant_records", "active_storage_blobs", column: "blob_id" + add_foreign_key "coplan_agent_events", "coplan_api_tokens", column: "api_token_id" + add_foreign_key "coplan_agent_events", "coplan_plans", column: "plan_id" + add_foreign_key "coplan_agent_sessions", "coplan_api_tokens", column: "api_token_id" + add_foreign_key "coplan_agent_sessions", "coplan_plans", column: "plan_id" add_foreign_key "coplan_api_tokens", "coplan_users", column: "user_id" add_foreign_key "coplan_comment_threads", "coplan_plan_versions", column: "addressed_in_plan_version_id" add_foreign_key "coplan_comment_threads", "coplan_plan_versions", column: "out_of_date_since_version_id" diff --git a/engine/app/assets/stylesheets/coplan/application.css b/engine/app/assets/stylesheets/coplan/application.css index ad37943c..e4c66398 100644 --- a/engine/app/assets/stylesheets/coplan/application.css +++ b/engine/app/assets/stylesheets/coplan/application.css @@ -762,6 +762,105 @@ img, svg { white-space: nowrap; } +/* Agent presence pill — live state of an agent session on this plan. + Broadcast-replaced; must read at a glance as "the agent is engaged". */ +.plan-agent-sessions { + display: flex; + align-items: center; + gap: var(--space-sm); +} + +.agent-pill { + display: inline-flex; + align-items: center; + gap: 0.4rem; + padding: 0.25rem 0.65rem; + border-radius: 999px; + font-size: var(--text-sm); + font-weight: 500; + color: var(--color-text-muted); + background: color-mix(in srgb, var(--color-primary) 8%, var(--color-surface)); + border: 1px solid color-mix(in srgb, var(--color-primary) 25%, transparent); + white-space: nowrap; +} + +.agent-pill__icon { + flex-shrink: 0; + color: var(--color-primary); +} + +.agent-pill__dot { + width: 6px; + height: 6px; + border-radius: 50%; + background: var(--color-primary); + animation: agent-pill-pulse 1.4s ease-in-out infinite; +} + +.agent-pill--awaiting_input { + background: color-mix(in srgb, var(--color-warning, #b45309) 10%, var(--color-surface)); + border-color: color-mix(in srgb, var(--color-warning, #b45309) 30%, transparent); +} + +.agent-pill--awaiting_input .agent-pill__dot, +.agent-pill--awaiting_input .agent-pill__icon { + color: var(--color-warning, #b45309); + background: var(--color-warning, #b45309); + animation: none; +} + +.agent-pill--awaiting_input .agent-pill__icon { + background: none; +} + +@keyframes agent-pill-pulse { + 0%, 100% { opacity: 1; transform: scale(1); } + 50% { opacity: 0.35; transform: scale(0.8); } +} + +/* Live-edit diff flashes — applied by live_update_controller when a + broadcast revision lands: changed blocks tint-and-fade, inline + insertions get an wash, removed text lingers struck-through in a + then collapses. Green wash = new words, not "success". */ +.agent-flash-block { + animation: agent-flash-bg 2.4s ease-out; +} + +ins.agent-flash { + text-decoration: none; + border-radius: 2px; + animation: agent-flash-ins 2.4s ease-out; +} + +del.agent-flash { + text-decoration: line-through; + border-radius: 2px; + color: var(--color-text-muted); + animation: agent-flash-del 2.4s ease-out forwards; +} + +@keyframes agent-flash-bg { + 0% { background: color-mix(in srgb, var(--color-primary) 14%, transparent); } + 100% { background: transparent; } +} + +@keyframes agent-flash-ins { + 0% { background: color-mix(in srgb, var(--color-success) 30%, transparent); } + 100% { background: transparent; } +} + +@keyframes agent-flash-del { + 0% { background: color-mix(in srgb, var(--color-danger, #b91c1c) 22%, transparent); opacity: 1; } + 60% { background: transparent; opacity: 0.7; } + 100% { opacity: 0; font-size: 0; } +} + +@media (prefers-reduced-motion: reduce) { + .agent-pill__dot { animation: none; } + .agent-flash-block, ins.agent-flash { animation: none; } + del.agent-flash { animation: none; display: none; } +} + /* Flash messages */ .flash { padding: var(--space-md) var(--space-lg); diff --git a/engine/app/controllers/coplan/api/v1/agent_events_controller.rb b/engine/app/controllers/coplan/api/v1/agent_events_controller.rb new file mode 100644 index 00000000..d4f4fee0 --- /dev/null +++ b/engine/app/controllers/coplan/api/v1/agent_events_controller.rb @@ -0,0 +1,117 @@ +module CoPlan + module Api + module V1 + # The agent-facing event inbox. Agents (or their bridge daemons) on + # laptops behind NAT can't be pushed to, so delivery is pull-based + # with two modes on one endpoint: + # + # Long-poll (default): GET /api/v1/agent/events?wait=25&cursor= + # Returns as soon as an event is available, or after `wait` + # seconds with an empty list. Plain JSON, works with curl in a + # loop through any proxy. + # + # SSE: same URL with Accept: text/event-stream + # Holds the response open and streams events as they land, with + # heartbeat comments so intermediaries don't kill the socket. + # + # Cursoring: event ids are UUIDv7 (time-ordered), so `cursor` is + # simply the last event id the client has seen. Without a cursor you + # get unacked events, so a crashed client picks up where it left off. + # Delivery is at-least-once — ack after successful processing. + class AgentEventsController < BaseController + include ActionController::Live + + MAX_WAIT = 55 + SSE_LIFETIME = 5.minutes + POLL_INTERVAL = 0.5 + + def index + unless @api_token + render json: { error: "Agent events require token authentication" }, status: :forbidden + return + end + + if request.headers["Accept"].to_s.include?("text/event-stream") + stream_events + else + long_poll_events + end + end + + # POST /api/v1/agent/events/ack {"cursor": ""} + # Marks everything up to and including the cursor as processed. + def ack + unless @api_token + render json: { error: "Agent events require token authentication" }, status: :forbidden + return + end + + cursor = params[:cursor].to_s + if cursor.blank? + render json: { error: "cursor is required" }, status: :unprocessable_content + return + end + + acked = AgentEvent.for_token(@api_token).pending + .where("id <= ?", cursor) + .update_all(acked_at: Time.current) + + render json: { acked: acked } + end + + private + + def long_poll_events + wait = params[:wait].to_i.clamp(0, MAX_WAIT) + deadline = Time.current + wait + + loop do + events = fetch_events + if events.any? || Time.current >= deadline + render json: { + events: events.map(&:as_api_json), + cursor: events.last&.id || params[:cursor] + } + return + end + sleep POLL_INTERVAL + end + end + + def stream_events + response.headers["Content-Type"] = "text/event-stream" + response.headers["Cache-Control"] = "no-cache" + response.headers["X-Accel-Buffering"] = "no" + + cursor = params[:cursor].presence + deadline = Time.current + SSE_LIFETIME + last_heartbeat = Time.current + + while Time.current < deadline + events = fetch_events(cursor: cursor) + events.each do |event| + response.stream.write("id: #{event.id}\nevent: #{event.event_type}\ndata: #{event.as_api_json.to_json}\n\n") + cursor = event.id + end + + if Time.current - last_heartbeat > 15 + response.stream.write(": heartbeat\n\n") + last_heartbeat = Time.current + end + sleep POLL_INTERVAL + end + rescue IOError, ActionController::Live::ClientDisconnected + # Client went away — normal for a streaming endpoint. + ensure + response.stream.close + end + + def fetch_events(cursor: params[:cursor].presence) + scope = AgentEvent.for_token(@api_token) + scope = cursor ? scope.after(cursor) : scope.pending + scope.oldest_first.limit(100).to_a + end + end + end + end +end diff --git a/engine/app/controllers/coplan/api/v1/agent_sessions_controller.rb b/engine/app/controllers/coplan/api/v1/agent_sessions_controller.rb new file mode 100644 index 00000000..0e6a9b63 --- /dev/null +++ b/engine/app/controllers/coplan/api/v1/agent_sessions_controller.rb @@ -0,0 +1,79 @@ +module CoPlan + module Api + module V1 + # An agent claims a session on a plan to (a) subscribe its token to + # the plan's event inbox and (b) drive the presence pill humans see. + # + # POST /api/v1/plans/:plan_id/agent_session {"agent_name": "Claude"} + # PATCH /api/v1/plans/:plan_id/agent_session {"state": "active", "detail": "editing Rollout"} + # + # States: pending / active / awaiting_input / complete (stale is set + # by the server, not the agent). The etiquette (mirrored from + # Linear's agent guidelines): flip to `active` within seconds of a + # wake — that's the fast ack humans see — then work as slowly as you + # need; use `awaiting_input` when a question is blocking you; land on + # `complete` when your turn is done. + class AgentSessionsController < BaseController + before_action :set_plan + before_action :authorize_plan_access! + before_action :require_token! + + def create + session = AgentSession.find_or_initialize_by(plan_id: @plan.id, api_token_id: @api_token.id) + session.agent_name = params[:agent_name].presence || @api_token.agent_name.presence || @api_token.name + session.state = "active" + session.last_activity_at = Time.current + session.save! + session.broadcast_pill + + render json: session_json(session), status: :created + rescue ActiveRecord::RecordInvalid => e + render json: { error: e.message }, status: :unprocessable_content + end + + def update + session = AgentSession.find_by(plan_id: @plan.id, api_token_id: @api_token.id) + unless session + render json: { error: "No agent session on this plan — POST to create one" }, status: :not_found + return + end + + state = params[:state].to_s + unless AgentSession::STATES.include?(state) && state != "stale" + render json: { error: "state must be one of #{(AgentSession::STATES - ['stale']).join(', ')}" }, status: :unprocessable_content + return + end + + session.transition!(state, detail: params[:detail].presence) + render json: session_json(session) + end + + def destroy + session = AgentSession.find_by(plan_id: @plan.id, api_token_id: @api_token.id) + session&.transition!("complete") + session&.destroy! + head :no_content + end + + private + + def require_token! + return if @api_token + + render json: { error: "Agent sessions require token authentication" }, status: :forbidden + end + + def session_json(session) + { + id: session.id, + plan_id: session.plan_id, + agent_name: session.agent_name, + state: session.state, + state_detail: session.state_detail, + last_activity_at: session.last_activity_at + } + end + end + end + end +end diff --git a/engine/app/controllers/coplan/api/v1/comments_controller.rb b/engine/app/controllers/coplan/api/v1/comments_controller.rb index cb58b5d1..89201a58 100644 --- a/engine/app/controllers/coplan/api/v1/comments_controller.rb +++ b/engine/app/controllers/coplan/api/v1/comments_controller.rb @@ -5,14 +5,33 @@ class CommentsController < BaseController before_action :set_plan before_action :authorize_plan_access! + # GET /api/v1/plans/:plan_id/comments/:id — a single thread with its + # comments, so an agent reacting to one inbox event doesn't have to + # refetch every thread on the plan. + def show + thread = @plan.comment_threads.includes(:comments, :created_by_user).find_by(id: params[:id]) + unless thread + render json: { error: "Comment thread not found" }, status: :not_found + return + end + + render json: thread_json(thread) + end + def create + # Same initial-status rule as the web flow: the plan author's own + # comments start as "todo" (self-assigned), everyone else's as + # "pending" (awaiting author triage). + initial_status = current_user&.id == @plan.created_by_user_id ? "todo" : "pending" + thread = @plan.comment_threads.new( plan_version: @plan.current_plan_version, anchor_text: params[:anchor_text].presence, anchor_occurrence: params[:anchor_occurrence]&.to_i, start_line: params[:start_line].presence, end_line: params[:end_line].presence, - created_by_user: current_user + created_by_user: current_user, + status: initial_status ) thread.save! @@ -156,6 +175,33 @@ def reply private + # Mirrors PlansController#thread_json so GET .../comments/:id returns + # the same shape as the list/snapshot endpoints. + def thread_json(thread) + { + id: thread.id, + status: thread.status, + anchor_text: thread.anchor_text, + anchor_context: thread.anchor_context_with_highlight, + anchor_valid: thread.anchor_valid?, + start_line: thread.start_line, + end_line: thread.end_line, + out_of_date: thread.out_of_date, + created_by: thread.created_by_user&.name, + created_at: thread.created_at, + comments: thread.comments.sort_by(&:created_at).map { |c| + { + id: c.id, + author_type: c.author_type, + author_id: c.author_id, + agent_name: c.agent_name, + body_markdown: c.body_markdown, + created_at: c.created_at + } + } + } + end + def broadcast_new_thread(thread) Broadcaster.append_to( @plan, diff --git a/engine/app/javascript/controllers/coplan/live_update_controller.js b/engine/app/javascript/controllers/coplan/live_update_controller.js index 924cb999..878fb2dc 100644 --- a/engine/app/javascript/controllers/coplan/live_update_controller.js +++ b/engine/app/javascript/controllers/coplan/live_update_controller.js @@ -8,9 +8,11 @@ import { Controller } from "@hotwired/stimulus" * (or anyone) commits a new revision elsewhere, the server pushes the new * rendered body to every open tab. This controller decides what to do: * - * * If the user has no unsaved drafts → swap the body in place. Existing - * Stimulus controllers reconnect over the new DOM, comment highlights - * re-attach. + * * If the user has no unsaved drafts → swap the body in place, then + * flash the sections this revision touched (the broadcast carries + * their keys in data-changed-sections): changed blocks tint-and-fade, + * and plain-text paragraphs get word-level / flashes so you + * can literally watch the agent's edit land. * * * If the user is mid-edit (any textarea on the page has non-empty, * non-trim-blank text) → DON'T blow away their typing. Instead, show @@ -51,6 +53,11 @@ export default class extends Controller { const currentRevision = parseInt(target.getAttribute("data-coplan--live-update-revision-value"), 10) || 0 if (incomingRevision && currentRevision >= incomingRevision) return + let changedKeys = [] + try { + changedKeys = JSON.parse(this.getAttribute("data-changed-sections") || "[]") + } catch { /* malformed attribute — fall back to a flash-free swap */ } + // `templateContent` is a DocumentFragment — it has no `innerHTML`. // Use replaceChildren(fragment) to swap the contents of target in one // shot. Stimulus controllers inside target will disconnect + reconnect. @@ -59,11 +66,13 @@ export default class extends Controller { if (hasDirtyDrafts()) { showStaleBanner(target, incomingRevision) } else { + const oldSections = snapshotSections(target, changedKeys) target.replaceChildren(fragment) if (incomingRevision) { target.setAttribute("data-coplan--live-update-revision-value", String(incomingRevision)) } clearStaleBanner() + if (changedKeys.length > 0) flashChangedSections(target, changedKeys, oldSections) } } @@ -89,6 +98,167 @@ function hasDirtyDrafts() { return false } +/* --------------------------------------------------------------------- + * Diff flashes. Section boundaries and slugs use the exact walk the + * changed-sections and TOC controllers use (top-level h1–h3 children of + * .markdown-rendered, slugified text, -2/-3 duplicate suffixes), so + * server keys and client sections stay in lockstep. + */ +const TOP_KEY = "__top__" +const FLASH_MS = 2600 +const MAX_DIFF_TOKENS = 600 + +function slugify(text, used) { + let base = text + .toLowerCase() + .replace(/\s+/g, "-") + .replace(/[^a-z0-9-]/g, "") + .replace(/-{2,}/g, "-") + .replace(/^-|-$/g, "") + if (base === "") base = "section" + let slug = base + let suffix = 2 + while (used.has(slug)) slug = `${base}-${suffix++}` + used.add(slug) + return slug +} + +// Map of section key → array of block elements (live nodes for the new +// DOM; for snapshots we keep outerHTML/text copies instead). +function sectionBlocks(root) { + const rendered = root.querySelector(".markdown-rendered") + if (!rendered) return new Map() + + const sections = new Map([[TOP_KEY, []]]) + const used = new Set() + let currentKey = TOP_KEY + + for (const node of Array.from(rendered.children)) { + if (/^H[1-3]$/.test(node.tagName)) { + currentKey = slugify(node.textContent, used) + sections.set(currentKey, []) + } + sections.get(currentKey).push(node) + } + return sections +} + +function snapshotSections(root, keys) { + if (!keys || keys.length === 0) return new Map() + const wanted = new Set(keys) + const snapshot = new Map() + for (const [key, blocks] of sectionBlocks(root)) { + if (!wanted.has(key)) continue + snapshot.set(key, blocks.map((el) => ({ + tag: el.tagName, + html: el.outerHTML, + text: el.textContent, + plainText: el.childElementCount === 0 + }))) + } + return snapshot +} + +function flashChangedSections(root, keys, oldSections) { + const wanted = new Set(keys) + const flashed = [] + + for (const [key, blocks] of sectionBlocks(root)) { + if (!wanted.has(key)) continue + const oldBlocks = oldSections.get(key) || [] + const oldHtml = new Set(oldBlocks.map((b) => b.html)) + + blocks.forEach((block, i) => { + if (oldHtml.has(block.outerHTML)) return // block untouched + + const old = oldBlocks[i] + if ( + old && old.tag === block.tagName && old.plainText && + block.childElementCount === 0 && blocks.length === oldBlocks.length + ) { + renderWordFlash(block, old.text, block.textContent) + } else { + block.classList.add("agent-flash-block") + } + flashed.push(block) + }) + } + + if (flashed.length > 0) { + flashed[0].scrollIntoView({ behavior: "smooth", block: "nearest" }) + setTimeout(() => settleFlashes(root), FLASH_MS) + } +} + +// Word-level flash for a plain-text block: rebuild its text as a +// common/inserted/deleted word sequence with / wrappers. The +// wrappers are temporary — settleFlashes() strips them — so comment +// anchors and copy/paste see clean text again within a couple seconds. +function renderWordFlash(block, oldText, newText) { + const oldTokens = tokenize(oldText) + const newTokens = tokenize(newText) + if (oldTokens.length > MAX_DIFF_TOKENS || newTokens.length > MAX_DIFF_TOKENS) { + block.classList.add("agent-flash-block") + return + } + + block.replaceChildren() + for (const part of diffTokens(oldTokens, newTokens)) { + if (part.type === "same") { + block.appendChild(document.createTextNode(part.text)) + } else { + const el = document.createElement(part.type === "ins" ? "ins" : "del") + el.className = "agent-flash" + el.textContent = part.text + block.appendChild(el) + } + } +} + +function tokenize(text) { + return text.split(/(\s+)/).filter((t) => t.length > 0) +} + +// Plain LCS word diff — paragraphs are small, O(n·m) is nothing, and it +// keeps the page dependency-free (no bundler, importmap-only app). +function diffTokens(a, b) { + const n = a.length, m = b.length + const lcs = Array.from({ length: n + 1 }, () => new Uint16Array(m + 1)) + for (let i = n - 1; i >= 0; i--) { + for (let j = m - 1; j >= 0; j--) { + lcs[i][j] = a[i] === b[j] ? lcs[i + 1][j + 1] + 1 : Math.max(lcs[i + 1][j], lcs[i][j + 1]) + } + } + + const parts = [] + const push = (type, text) => { + const last = parts[parts.length - 1] + if (last && last.type === type) last.text += text + else parts.push({ type, text }) + } + + let i = 0, j = 0 + while (i < n && j < m) { + if (a[i] === b[j]) { push("same", a[i]); i++; j++ } + else if (lcs[i + 1][j] >= lcs[i][j + 1]) { push("del", a[i]); i++ } + else { push("ins", b[j]); j++ } + } + while (i < n) { push("del", a[i]); i++ } + while (j < m) { push("ins", b[j]); j++ } + return parts +} + +function settleFlashes(root) { + for (const del of root.querySelectorAll("del.agent-flash")) del.remove() + for (const ins of root.querySelectorAll("ins.agent-flash")) { + ins.replaceWith(document.createTextNode(ins.textContent)) + } + for (const block of root.querySelectorAll(".agent-flash-block")) { + block.classList.remove("agent-flash-block") + } + root.normalize() +} + function showStaleBanner(targetEl, revision) { let banner = document.getElementById("plan-stale-banner") if (banner) { diff --git a/engine/app/jobs/coplan/mark_stale_agent_session_job.rb b/engine/app/jobs/coplan/mark_stale_agent_session_job.rb new file mode 100644 index 00000000..67aeab4d --- /dev/null +++ b/engine/app/jobs/coplan/mark_stale_agent_session_job.rb @@ -0,0 +1,18 @@ +module CoPlan + # Scheduled at wake time (AgentSession#wake!). If the session hasn't + # emitted any activity since that wake, the pill would be a lie — flip + # it to stale so the UI stops showing "on it…" for an agent that never + # showed up. + class MarkStaleAgentSessionJob < ApplicationJob + def perform(agent_session_id:, woken_at:) + session = AgentSession.find_by(id: agent_session_id) + return unless session + return unless session.state == "pending" + + # Any activity after the wake means the agent reacted; leave it be. + return if session.last_activity_at.present? && session.last_activity_at > Time.iso8601(woken_at) + + session.transition!("stale") + end + end +end diff --git a/engine/app/models/coplan/agent_event.rb b/engine/app/models/coplan/agent_event.rb new file mode 100644 index 00000000..9112f746 --- /dev/null +++ b/engine/app/models/coplan/agent_event.rb @@ -0,0 +1,36 @@ +module CoPlan + # One row per (event, subscribed agent token) — the durable inbox an agent + # (or its bridge daemon) drains via GET /api/v1/agent/events. IDs are + # UUIDv7, so lexicographic order is creation order and the id doubles as + # the resume cursor. + class AgentEvent < ApplicationRecord + TYPES = %w[ + comment.created + comment.replied + thread.status_changed + plan.content_changed + ].freeze + + belongs_to :api_token, class_name: "CoPlan::ApiToken" + belongs_to :plan, class_name: "CoPlan::Plan" + + validates :event_type, inclusion: { in: TYPES } + + scope :for_token, ->(token) { where(api_token_id: token.id) } + scope :pending, -> { where(acked_at: nil) } + scope :after, ->(cursor) { where("id > ?", cursor) } + scope :oldest_first, -> { order(:id) } + + def as_api_json + { + id: id, + type: event_type, + plan_id: plan_id, + comment_thread_id: comment_thread_id, + comment_id: comment_id, + payload: payload, + created_at: created_at + } + end + end +end diff --git a/engine/app/models/coplan/agent_session.rb b/engine/app/models/coplan/agent_session.rb new file mode 100644 index 00000000..f9dca8d9 --- /dev/null +++ b/engine/app/models/coplan/agent_session.rb @@ -0,0 +1,63 @@ +module CoPlan + # A delegation of a plan to an agent — one per (plan, api_token). The + # state machine mirrors Linear's agent sessions: it drives the presence + # pill on the plan page, so humans can see the agent is engaged the + # moment it wakes, and whose turn it is when it asks a question. + # + # pending an event was published; the agent hasn't reacted yet + # active the agent acked / is working (state_detail says what) + # awaiting_input the agent asked a question; it's the human's turn + # complete the agent finished its turn + # stale the agent never reacted to a wake — don't show a + # zombie "typing…" pill + class AgentSession < ApplicationRecord + STATES = %w[pending active awaiting_input complete stale].freeze + + # States rendered as a live pill on the plan page. + VISIBLE_STATES = %w[pending active awaiting_input].freeze + + # How long a session may sit in `pending` after a wake before we stop + # promising the humans that the agent is coming. + STALE_AFTER = 30.seconds + + belongs_to :plan, class_name: "CoPlan::Plan" + belongs_to :api_token, class_name: "CoPlan::ApiToken" + + validates :agent_name, presence: true + validates :state, inclusion: { in: STATES } + + scope :visible, -> { where(state: VISIBLE_STATES) } + + def transition!(new_state, detail: nil) + update!(state: new_state, state_detail: detail, last_activity_at: Time.current) + broadcast_pill + end + + # Called when an event is published to this session's inbox: flip back + # to pending (unless the agent is already mid-turn) and start the + # stale countdown. + def wake! + transition!("pending") unless state == "active" + MarkStaleAgentSessionJob.set(wait: STALE_AFTER).perform_later( + agent_session_id: id, woken_at: last_activity_at&.iso8601 || Time.current.iso8601 + ) + end + + def display_status + case state + when "pending" then "#{agent_name} is on it…" + when "active" then state_detail.presence ? "#{agent_name} is #{state_detail}" : "#{agent_name} is working…" + when "awaiting_input" then "#{agent_name} asked a question" + end + end + + def broadcast_pill + Broadcaster.replace_to( + plan, + target: "plan-agent-sessions", + partial: "coplan/plans/agent_sessions", + locals: { agent_sessions: AgentSession.visible.where(plan_id: plan_id).order(:created_at) } + ) + end + end +end diff --git a/engine/app/services/coplan/agent_events/publish.rb b/engine/app/services/coplan/agent_events/publish.rb new file mode 100644 index 00000000..05076d1d --- /dev/null +++ b/engine/app/services/coplan/agent_events/publish.rb @@ -0,0 +1,58 @@ +module CoPlan + module AgentEvents + # Fans an event out to the inbox of every agent session on the plan, + # except the actor's own (an agent should not be woken by its own + # comment or edit). Suppression matches on the api token id, which is + # what api_actor_id returns for token-authenticated callers. + class Publish + def self.call(plan:, event_type:, actor_id: nil, comment_thread: nil, comment: nil, payload: {}) + new(plan:, event_type:, actor_id:, comment_thread:, comment:, payload:).call + end + + def initialize(plan:, event_type:, actor_id:, comment_thread:, comment:, payload:) + @plan = plan + @event_type = event_type + @actor_id = actor_id + @comment_thread = comment_thread + @comment = comment + @payload = payload + end + + def call + sessions = AgentSession.where(plan_id: @plan.id) + sessions.each do |session| + next if @actor_id.present? && session.api_token_id == @actor_id + + AgentEvent.create!( + api_token_id: session.api_token_id, + plan_id: @plan.id, + comment_thread_id: @comment_thread&.id, + comment_id: @comment&.id, + event_type: @event_type, + payload: base_payload.merge(@payload) + ) + session.wake! + end + end + + private + + def base_payload + payload = { + "plan_title" => @plan.title, + "plan_revision" => @plan.current_revision + } + if @comment_thread + payload["thread_status"] = @comment_thread.status + payload["anchor_text"] = @comment_thread.anchor_text + end + if @comment + payload["comment_body"] = @comment.body_markdown + payload["comment_author"] = @comment.agent_name.presence || @comment.author&.name + payload["comment_author_type"] = @comment.author_type + end + payload + end + end + end +end diff --git a/engine/app/services/coplan/broadcaster.rb b/engine/app/services/coplan/broadcaster.rb index 5e7f2c22..6551980d 100644 --- a/engine/app/services/coplan/broadcaster.rb +++ b/engine/app/services/coplan/broadcaster.rb @@ -48,14 +48,20 @@ def custom_action_to(streamable, action:, target:, html:, attrs: {}) # push the freshly rendered plan body to every open tab, letting the # client decide whether to apply it (clean) or show a stale-revision # banner (dirty draft in progress). - def replace_plan_content(plan) + # + # changed_sections (section keys from Plans::ChangedSections) rides + # along so viewing tabs can flash exactly the sections this revision + # touched instead of hard-swapping the whole body silently. + def replace_plan_content(plan, changed_sections: nil) html = render(partial: "coplan/plans/content_body", locals: { plan: plan }) + attrs = { "data-revision" => plan.current_revision } + attrs["data-changed-sections"] = changed_sections.to_json if changed_sections.present? custom_action_to( plan, action: "coplan-replace-if-clean", target: "plan-content-body", html: html, - attrs: { "data-revision" => plan.current_revision } + attrs: attrs ) end diff --git a/engine/app/services/coplan/notifications/create.rb b/engine/app/services/coplan/notifications/create.rb index d1f932da..b8a9b657 100644 --- a/engine/app/services/coplan/notifications/create.rb +++ b/engine/app/services/coplan/notifications/create.rb @@ -13,6 +13,8 @@ def initialize(comment_thread:, actor_id:, comment: nil, reason:) end def call + publish_agent_events + subscriber_ids = compute_subscribers subscriber_ids.delete(@actor_id) return if subscriber_ids.empty? @@ -32,6 +34,30 @@ def call private + # Every comment/thread write path already funnels through this + # service, so it doubles as the choke point for the agent event + # inbox. Human notification fan-out below is unchanged; agents get + # their own recipients (sessions on the plan), and @actor_id — the + # api token id for token-authenticated callers — suppresses an + # agent being woken by its own activity. + def publish_agent_events + event_type = + case @reason + when "new_comment" then "comment.created" + when "reply", "agent_response" then "comment.replied" + when "status_change" then "thread.status_changed" + end + return unless event_type + + AgentEvents::Publish.call( + plan: @comment_thread.plan, + event_type: event_type, + actor_id: @actor_id, + comment_thread: @comment_thread, + comment: @comment + ) + end + def compute_subscribers case @reason when "new_comment" diff --git a/engine/app/services/coplan/plans/replace_content.rb b/engine/app/services/coplan/plans/replace_content.rb index 07f17cb4..68ba58ee 100644 --- a/engine/app/services/coplan/plans/replace_content.rb +++ b/engine/app/services/coplan/plans/replace_content.rb @@ -121,7 +121,22 @@ def call partial: "coplan/plans/header", locals: { plan: @plan } ) - Broadcaster.replace_plan_content(@plan) + changed_sections = Plans::ChangedSections.call( + old_content: current_content, + new_content: @new_content + ) + Broadcaster.replace_plan_content(@plan, changed_sections: changed_sections) + + AgentEvents::Publish.call( + plan: @plan, + event_type: "plan.content_changed", + actor_id: @actor_id, + payload: { + "revision" => new_revision, + "change_summary" => @change_summary, + "changed_sections" => changed_sections + } + ) { version: version, plan: @plan, applied: result[:applied].length, no_op: false } end diff --git a/engine/app/views/coplan/agent_instructions/show.text.erb b/engine/app/views/coplan/agent_instructions/show.text.erb index 39b3a70b..1859a320 100644 --- a/engine/app/views/coplan/agent_instructions/show.text.erb +++ b/engine/app/views/coplan/agent_instructions/show.text.erb @@ -499,13 +499,91 @@ Mark a comment thread as resolved (addressed by the plan author or thread creato ### Dismiss Thread -Dismiss a comment thread (plan author only — for comments that are out of scope or not applicable). +Dismiss a comment thread (plan author only — for comments that are out of scope or not applicable). `discard` is accepted as an alias for `dismiss`. ```bash <%= @curl %> -X PATCH \ "<%= @base %>/api/v1/plans/$PLAN_ID/comments/$THREAD_ID/dismiss" | jq . ``` +### Get a Single Thread + +Fetch one thread with its comments — handy when reacting to a single event without refetching the whole snapshot. + +```bash +<%= @curl %> "<%= @base %>/api/v1/plans/$PLAN_ID/comments/$THREAD_ID" | jq . +``` + +## Live Collaboration (Realtime) + +If you authored a plan (or are babysitting one), you can hear about new comments and edits the moment they happen instead of re-polling snapshots, and show humans that you're on it. + +### 1. Claim an Agent Session + +Claiming a session subscribes your token to the plan's events and shows a live presence pill ("Claude is working…") to everyone viewing the plan. + +```bash +<%= @curl %> -X POST \ + -H "Content-Type: application/json" \ + -d '{"agent_name": "Claude"}' \ + "<%= @base %>/api/v1/plans/$PLAN_ID/agent_session" | jq . +``` + +Keep the pill honest by updating your state as you work (`state`: `pending`, `active`, `awaiting_input`, `complete`; optional `detail` shows what you're doing): + +```bash +<%= @curl %> -X PATCH \ + -H "Content-Type: application/json" \ + -d '{"state": "active", "detail": "tightening the Rollout section"}' \ + "<%= @base %>/api/v1/plans/$PLAN_ID/agent_session" | jq . +``` + +Etiquette (this is what makes you feel like a collaborator, not a batch job): + +- Flip to `active` within a few seconds of waking — the fast ack humans see. Work as slowly as you need after that; if you stay silent in `pending` the pill goes stale. +- Post a short reply narrating what you're about to change *before* you PUT the edit. +- Use `awaiting_input` when a question blocks you; land on `complete` when your turn is done. +- `DELETE` the session when you stop watching the plan. + +### 2. Wait for Events + +Long-poll (returns immediately when an event lands, or empty after `wait` seconds): + +```bash +<%= @curl %> "<%= @base %>/api/v1/agent/events?wait=25&cursor=$LAST_EVENT_ID" | jq . +``` + +Event types: `comment.created`, `comment.replied`, `thread.status_changed`, `plan.content_changed`. Each event carries `plan_id`, `comment_thread_id`, and a payload with the comment body/anchor so you usually don't need another fetch to decide what to do. Your own comments and edits are never echoed back to you. + +Cursoring: event ids are time-ordered — pass the last id you processed as `cursor` to resume. Without a cursor you get all unacked events. Delivery is at-least-once; ack after processing: + +```bash +<%= @curl %> -X POST \ + -H "Content-Type: application/json" \ + -d '{"cursor": "$LAST_EVENT_ID"}' \ + "<%= @base %>/api/v1/agent/events/ack" | jq . +``` + +Prefer a held-open stream? The same endpoint speaks Server-Sent Events with `-H "Accept: text/event-stream"` (heartbeats every 15s, reconnect with your cursor). + +### 3. The Loop + +```bash +CURSOR="" +while true; do + RESP=$(<%= @curl %> "<%= @base %>/api/v1/agent/events?wait=25&cursor=$CURSOR") + echo "$RESP" | jq -c '.events[]' | while read -r event; do + # 1. PATCH agent_session state=active (fast ack — do this first) + # 2. Read the event payload / thread, decide, reply, edit + # 3. PATCH agent_session state=complete + : + done + NEXT=$(echo "$RESP" | jq -r '.cursor // empty') + [ -n "$NEXT" ] && CURSOR="$NEXT" && <%= @curl %> -X POST -H "Content-Type: application/json" \ + -d "{\"cursor\": \"$CURSOR\"}" "<%= @base %>/api/v1/agent/events/ack" > /dev/null +done +``` + ## Reviewing a Plan When asked to review a plan (given a plan URL or ID), follow this workflow: diff --git a/engine/app/views/coplan/plans/_agent_sessions.html.erb b/engine/app/views/coplan/plans/_agent_sessions.html.erb new file mode 100644 index 00000000..600066f4 --- /dev/null +++ b/engine/app/views/coplan/plans/_agent_sessions.html.erb @@ -0,0 +1,13 @@ +<%# The agent presence pill(s): live state of every agent session on this + plan (pending / active / awaiting_input). Broadcast-replaced whole by + AgentSession#broadcast_pill, so keep it requestless-renderable — no + forms, no current_user. %> +
+ <% agent_sessions.each do |session| %> +
+ + + <%= session.display_status %> +
+ <% end %> +
diff --git a/engine/app/views/coplan/plans/show.html.erb b/engine/app/views/coplan/plans/show.html.erb index 55678405..cf74bb92 100644 --- a/engine/app/views/coplan/plans/show.html.erb +++ b/engine/app/views/coplan/plans/show.html.erb @@ -38,6 +38,7 @@
<%= render partial: "coplan/plans/header", locals: { plan: @plan } %>
+ <%= render partial: "coplan/plans/agent_sessions", locals: { agent_sessions: CoPlan::AgentSession.visible.where(plan_id: @plan.id).order(:created_at) } %> <%= render partial: "coplan/plans/viewers", locals: { viewers: CoPlan::PlanViewer.active_viewers_for(@plan), current_user: current_user } %> <%= render partial: "coplan/plans/toolbar", locals: { plan: @plan, my_placement: my_placement } %>
diff --git a/engine/config/routes.rb b/engine/config/routes.rb index eacff866..0485ce0d 100644 --- a/engine/config/routes.rb +++ b/engine/config/routes.rb @@ -61,11 +61,18 @@ resources :sessions, only: [:create, :show], controller: "sessions" do post :commit, on: :member end - resources :comments, only: [:create], controller: "comments" do + resources :comments, only: [:create, :show], controller: "comments" do post :reply, on: :member patch :resolve, on: :member patch :discard, on: :member + # Alias: the agent instructions long documented this action as + # "dismiss" while the route said "discard" — accept both so + # agents following either name succeed. + patch :dismiss, on: :member, action: :discard end + # Presence/state for an agent working this plan (drives the + # "Claude is editing…" pill and subscribes the token to events). + resource :agent_session, only: [:create, :update, :destroy], controller: "agent_sessions" # Deletes an individual comment (by comment ID, not thread ID). # Distinct from the routes above, which key off thread ID. delete "comments/:id/delete", to: "comments#destroy", as: :destroy_comment @@ -75,6 +82,13 @@ resources :references, only: [] do get :search, on: :collection end + + # Agent event inbox — pull-based (long-poll or SSE) so agents on + # laptops behind NAT can hear about comments the moment they land. + scope :agent do + get "events", to: "agent_events#index", as: :agent_events + post "events/ack", to: "agent_events#ack", as: :agent_events_ack + end end end diff --git a/engine/db/migrate/20260807000000_create_agent_collaboration_tables.rb b/engine/db/migrate/20260807000000_create_agent_collaboration_tables.rb new file mode 100644 index 00000000..cb7d7014 --- /dev/null +++ b/engine/db/migrate/20260807000000_create_agent_collaboration_tables.rb @@ -0,0 +1,45 @@ +class CreateAgentCollaborationTables < ActiveRecord::Migration[8.1] + def change + # Durable per-agent event inbox. IDs are UUIDv7 (time-ordered), so the + # id doubles as the pagination cursor: "give me events after ". + create_table :coplan_agent_events, id: { type: :string, limit: 36 } do |t| + t.string :api_token_id, limit: 36, null: false + t.string :plan_id, limit: 36, null: false + t.string :comment_thread_id, limit: 36 + t.string :comment_id, limit: 36 + t.string :event_type, null: false + t.json :payload + t.datetime :acked_at + t.datetime :created_at, null: false + + t.index [:api_token_id, :id] + t.index [:api_token_id, :acked_at] + t.index :plan_id + end + + # One session per (plan, agent token) — Linear-style delegation state + # machine driving the presence pill: pending / active / awaiting_input / + # complete / stale. + create_table :coplan_agent_sessions, id: { type: :string, limit: 36 } do |t| + t.string :plan_id, limit: 36, null: false + t.string :api_token_id, limit: 36, null: false + t.string :agent_name, null: false + t.string :state, null: false, default: "pending" + t.string :state_detail + t.datetime :last_activity_at + t.timestamps + + t.index [:plan_id, :api_token_id], unique: true + t.index :api_token_id + end + + # Stable display identity for an agent token, instead of the free-text + # per-comment agent_name. + add_column :coplan_api_tokens, :agent_name, :string + + add_foreign_key :coplan_agent_events, :coplan_api_tokens, column: :api_token_id + add_foreign_key :coplan_agent_events, :coplan_plans, column: :plan_id + add_foreign_key :coplan_agent_sessions, :coplan_api_tokens, column: :api_token_id + add_foreign_key :coplan_agent_sessions, :coplan_plans, column: :plan_id + end +end diff --git a/script/coplan-bridge b/script/coplan-bridge new file mode 100644 index 00000000..62bc3704 --- /dev/null +++ b/script/coplan-bridge @@ -0,0 +1,199 @@ +#!/usr/bin/env ruby +# frozen_string_literal: true + +# coplan-bridge — wakes your local coding agent when someone comments on a +# CoPlan plan. Harness-agnostic: it drains the plan's agent event inbox +# (long-poll, NAT-proof) and injects each event into whichever harness you +# run via a per-harness "resume session with a message" command. +# +# script/coplan-bridge --config bridge.json +# +# Config (JSON): +# { +# "base_url": "http://localhost:3000", +# "token": "", +# "agent_name": "Claude", +# "plans": ["", ...], // sessions are claimed on start +# "adapter": "claude", // which command template to use +# "harness_session": "", +# "adapters": { ... } // optional overrides, see DEFAULT_ADAPTERS +# } +# +# The adapter command receives the rendered wake prompt as its final +# argument ({prompt}) and the harness session id as {session}. The prompt +# tells the agent to do the polite thing: fast-ack the session, reply, +# edit, complete. The bridge itself flips the session pill to `active` +# before the harness even boots, so humans see life within a second. +# +# The "demo" adapter needs no harness or API budget: it acks, replies to +# the thread, and makes a small visible edit — enough to exercise the +# full live-collaboration UI end to end. + +require "json" +require "net/http" +require "uri" +require "optparse" + +DEFAULT_ADAPTERS = { + # Every entry is "resume a named session with this message" in that + # harness's dialect. {session} and {prompt} are substituted. + "claude" => ["claude", "-p", "--resume", "{session}", "--permission-mode", "acceptEdits", "{prompt}"], + "codex" => ["codex", "exec", "resume", "{session}", "{prompt}"], + "goose" => ["goose", "run", "--name", "{session}", "--resume", "-t", "{prompt}"], + "openhands" => ["openhands", "--headless", "--resume", "{session}", "-t", "{prompt}"], + "amp" => ["amp", "threads", "continue", "{session}", "-x", "{prompt}"], + "demo" => [] # handled in-process, see DemoAgent +}.freeze + +options = { config: File.expand_path("~/.config/coplan/bridge.json") } +OptionParser.new do |opts| + opts.banner = "Usage: coplan-bridge [--config PATH]" + opts.on("--config PATH", "Path to bridge config JSON") { |v| options[:config] = v } +end.parse! + +CONFIG = JSON.parse(File.read(options[:config])) +BASE = URI(CONFIG.fetch("base_url")) +TOKEN = CONFIG.fetch("token") + +def request(method, path, body: nil, params: nil, read_timeout: 90) + uri = URI.join(BASE.to_s, path) + uri.query = URI.encode_www_form(params) if params + klass = { get: Net::HTTP::Get, post: Net::HTTP::Post, patch: Net::HTTP::Patch, put: Net::HTTP::Put }.fetch(method) + req = klass.new(uri) + req["Authorization"] = "Bearer #{TOKEN}" + req["Content-Type"] = "application/json" + req.body = body.to_json if body + Net::HTTP.start(uri.hostname, uri.port, use_ssl: uri.scheme == "https", read_timeout: read_timeout) do |http| + res = http.request(req) + warn "coplan-bridge: #{method.upcase} #{path} -> #{res.code}" unless res.code.start_with?("2") + res.body && !res.body.empty? ? JSON.parse(res.body) : {} + end +rescue JSON::ParserError + {} +end + +def claim_sessions + Array(CONFIG["plans"]).each do |plan_id| + request(:post, "/api/v1/plans/#{plan_id}/agent_session", body: { agent_name: CONFIG.fetch("agent_name", "Agent") }) + puts "coplan-bridge: watching plan #{plan_id}" + end +end + +def wake_prompt(event) + payload = event["payload"] || {} + <<~PROMPT + [coplan-bridge] A CoPlan event needs your attention. + + Event: #{event["type"]} on plan "#{payload["plan_title"]}" (plan id #{event["plan_id"]}, thread id #{event["comment_thread_id"]}). + #{payload["comment_author"] ? "#{payload["comment_author"]} wrote: #{payload["comment_body"].to_s.strip}" : ""} + #{payload["anchor_text"] ? "Anchored to: \"#{payload["anchor_text"]}\"" : ""} + + Respond as the plan's agent, over the CoPlan API at #{BASE} (full instructions: #{BASE}/agent-instructions): + 1. PATCH /api/v1/plans/#{event["plan_id"]}/agent_session {"state":"active","detail":""} immediately. + 2. Reply on the thread first (narrate what you'll change), then apply any edit with PUT .../content. + 3. Finish with {"state":"complete"} (or "awaiting_input" if you asked a question). + PROMPT +end + +# A deterministic stand-in agent so the whole loop can be demoed without +# spending a real harness turn: ack -> narrated reply -> small visible edit. +class DemoAgent + def self.handle(event) + plan_id = event["plan_id"] + payload = event["payload"] || {} + return unless event["type"].start_with?("comment.") + + request_method = method(:request) + Bridge.patch_state(plan_id, "active", "rewording that section") + sleep 1 # visible thinking beat + + if event["comment_thread_id"] + Bridge.reply(plan_id, event["comment_thread_id"], "🤖 Got it — making that clearer now.") + end + + snapshot = Bridge.request(:get, "/api/v1/plans/#{plan_id}/snapshot") + content = snapshot["current_content"].to_s + anchor = payload["anchor_text"].to_s + if anchor.length > 3 && content.include?(anchor) + new_content = content.sub(anchor, rewrite(anchor)) + Bridge.request(:put, "/api/v1/plans/#{plan_id}/content", body: { + base_revision: snapshot["current_revision"], + content: new_content, + change_summary: "Reworded per comment (demo agent)" + }) + end + + Bridge.patch_state(plan_id, "complete", nil) + end + + # A visible, boring rewrite: swap stiff words for plain ones. + def self.rewrite(text) + text + .gsub(/\butilize\b/i, "use") + .gsub(/\bleverage\b/i, "use") + .gsub(/\bfacilitate\b/i, "help") + .gsub(/\bin order to\b/i, "to") + .gsub(/\badditionally\b/i, "also") + end +end + +module Bridge + module_function + + def request(method, path, body: nil, params: nil) + Object.send(:request, method, path, body: body, params: params) + end + + def patch_state(plan_id, state, detail) + request(:patch, "/api/v1/plans/#{plan_id}/agent_session", body: { state: state, detail: detail }.compact) + end + + def reply(plan_id, thread_id, text) + request(:post, "/api/v1/plans/#{plan_id}/comments/#{thread_id}/reply", body: { + body_markdown: text, agent_name: CONFIG.fetch("agent_name", "Agent") + }) + end +end + +def dispatch(event) + adapter = CONFIG.fetch("adapter", "demo") + if adapter == "demo" + DemoAgent.handle(event) + return + end + + template = (CONFIG["adapters"] || {}).fetch(adapter, DEFAULT_ADAPTERS[adapter]) + abort "coplan-bridge: unknown adapter #{adapter}" unless template + + # Honest pill before the harness even boots. + Bridge.patch_state(event["plan_id"], "active", "reading your comment") + + cmd = template.map do |part| + part.gsub("{session}", CONFIG.fetch("harness_session", "coplan")).gsub("{prompt}", wake_prompt(event)) + end + puts "coplan-bridge: dispatching #{event["type"]} -> #{adapter}" + system(*cmd) +end + +claim_sessions +cursor = nil +puts "coplan-bridge: draining events from #{BASE}" +loop do + resp = request(:get, "/api/v1/agent/events", params: { wait: 25, cursor: cursor }.compact) + events = resp["events"] || [] + events.each do |event| + dispatch(event) + rescue => e + warn "coplan-bridge: dispatch failed: #{e.class}: #{e.message}" + end + if (next_cursor = resp["cursor"]) && next_cursor != cursor + cursor = next_cursor + request(:post, "/api/v1/agent/events/ack", body: { cursor: cursor }) + end +rescue Interrupt + puts "\ncoplan-bridge: bye" + exit 0 +rescue => e + warn "coplan-bridge: #{e.class}: #{e.message} — retrying in 3s" + sleep 3 +end diff --git a/spec/requests/api/v1/agent_events_spec.rb b/spec/requests/api/v1/agent_events_spec.rb new file mode 100644 index 00000000..c8708fe8 --- /dev/null +++ b/spec/requests/api/v1/agent_events_spec.rb @@ -0,0 +1,165 @@ +require "rails_helper" + +RSpec.describe "Api::V1::AgentEvents", type: :request do + let(:hampton) { create(:coplan_user, :admin) } + let(:agent_token) { create(:api_token, user: hampton, raw_token: "test-token-agent", agent_name: "Claude") } + let(:agent_headers) { { "Authorization" => "Bearer test-token-agent" } } + let(:other_token) { create(:api_token, user: hampton, raw_token: "test-token-other", agent_name: "Amp") } + let(:other_headers) { { "Authorization" => "Bearer test-token-other" } } + let(:plan) { create(:plan, :considering, created_by_user: hampton) } + + before do + agent_token + other_token + end + + describe "agent sessions" do + it "claims a session, drives states, and rejects bogus states" do + post api_v1_plan_agent_session_path(plan), params: { agent_name: "Claude" }, headers: agent_headers, as: :json + expect(response).to have_http_status(:created) + body = JSON.parse(response.body) + expect(body["state"]).to eq("active") + expect(body["agent_name"]).to eq("Claude") + + patch api_v1_plan_agent_session_path(plan), params: { state: "awaiting_input", detail: "asked about rollout" }, headers: agent_headers, as: :json + expect(response).to have_http_status(:ok) + expect(JSON.parse(response.body)["state"]).to eq("awaiting_input") + + patch api_v1_plan_agent_session_path(plan), params: { state: "stale" }, headers: agent_headers, as: :json + expect(response).to have_http_status(:unprocessable_content) + end + + it "falls back to the token's agent_name when none is passed" do + post api_v1_plan_agent_session_path(plan), headers: agent_headers, as: :json + expect(JSON.parse(response.body)["agent_name"]).to eq("Claude") + end + + it "is idempotent per (plan, token)" do + 2.times { post api_v1_plan_agent_session_path(plan), headers: agent_headers, as: :json } + expect(CoPlan::AgentSession.where(plan_id: plan.id, api_token_id: agent_token.id).count).to eq(1) + end + end + + describe "event fan-out" do + let!(:session) { create_agent_collab_session(agent_token) } + let(:thread) { create(:comment_thread, plan: plan, plan_version: plan.current_plan_version, created_by_user: hampton) } + let(:comment) { create(:comment, comment_thread: thread, author_type: "human", author_id: hampton.id, body_markdown: "too formal!") } + + it "publishes comment events to subscribed agents and wakes the session" do + CoPlan::Notifications::Create.call(comment_thread: thread, actor_id: hampton.id, comment: comment, reason: "new_comment") + + events = CoPlan::AgentEvent.for_token(agent_token) + expect(events.count).to eq(1) + event = events.first + expect(event.event_type).to eq("comment.created") + expect(event.payload["comment_body"]).to eq("too formal!") + expect(event.payload["plan_title"]).to eq(plan.title) + + expect(session.reload.state).to eq("pending") + end + + it "does not wake the agent for its own comments" do + CoPlan::Notifications::Create.call(comment_thread: thread, actor_id: agent_token.id, comment: comment, reason: "agent_response") + + expect(CoPlan::AgentEvent.for_token(agent_token).count).to eq(0) + end + + it "publishes content changes with changed section keys" do + CoPlan::Plans::ReplaceContent.call( + plan: plan, + new_content: "# Title\n\nBrand new body.\n", + base_revision: plan.current_revision, + actor_type: "human", + actor_id: hampton.id, + change_summary: "Rewrite" + ) + + event = CoPlan::AgentEvent.for_token(agent_token).where(event_type: "plan.content_changed").first + expect(event).to be_present + expect(event.payload["changed_sections"]).to be_an(Array) + expect(event.payload["change_summary"]).to eq("Rewrite") + end + end + + describe "GET /api/v1/agent/events" do + let!(:session) { create_agent_collab_session(agent_token) } + let!(:other_session) { create_agent_collab_session(other_token) } + let(:thread) { create(:comment_thread, plan: plan, plan_version: plan.current_plan_version, created_by_user: hampton) } + + before do + comment = create(:comment, comment_thread: thread, author_type: "human", author_id: hampton.id, body_markdown: "hello agents") + CoPlan::Notifications::Create.call(comment_thread: thread, actor_id: hampton.id, comment: comment, reason: "new_comment") + end + + it "returns pending events for the calling token only" do + get api_v1_agent_events_path, params: { wait: 0 }, headers: agent_headers + expect(response).to have_http_status(:ok) + body = JSON.parse(response.body) + expect(body["events"].length).to eq(1) + expect(body["events"].first["type"]).to eq("comment.created") + expect(body["cursor"]).to eq(body["events"].first["id"]) + end + + it "resumes from a cursor" do + get api_v1_agent_events_path, params: { wait: 0 }, headers: agent_headers + cursor = JSON.parse(response.body)["cursor"] + + get api_v1_agent_events_path, params: { wait: 0, cursor: cursor }, headers: agent_headers + expect(JSON.parse(response.body)["events"]).to be_empty + + reply = create(:comment, comment_thread: thread, author_type: "human", author_id: hampton.id, body_markdown: "and another") + CoPlan::Notifications::Create.call(comment_thread: thread, actor_id: hampton.id, comment: reply, reason: "reply") + + get api_v1_agent_events_path, params: { wait: 0, cursor: cursor }, headers: agent_headers + events = JSON.parse(response.body)["events"] + expect(events.length).to eq(1) + expect(events.first["type"]).to eq("comment.replied") + end + + it "acks up to a cursor" do + get api_v1_agent_events_path, params: { wait: 0 }, headers: agent_headers + cursor = JSON.parse(response.body)["cursor"] + + post api_v1_agent_events_ack_path, params: { cursor: cursor }, headers: agent_headers, as: :json + expect(JSON.parse(response.body)["acked"]).to eq(1) + + get api_v1_agent_events_path, params: { wait: 0 }, headers: agent_headers + expect(JSON.parse(response.body)["events"]).to be_empty + end + end + + describe "comment API ergonomics" do + let(:thread) { create(:comment_thread, plan: plan, plan_version: plan.current_plan_version, created_by_user: hampton) } + + before { create(:comment, comment_thread: thread, author_type: "human", author_id: hampton.id, body_markdown: "first") } + + it "fetches a single thread" do + get api_v1_plan_comment_path(plan, thread), headers: agent_headers + expect(response).to have_http_status(:ok) + body = JSON.parse(response.body) + expect(body["id"]).to eq(thread.id) + expect(body["comments"].length).to eq(1) + end + + it "accepts dismiss as an alias for discard" do + patch dismiss_api_v1_plan_comment_path(plan, thread), headers: agent_headers, as: :json + expect(response).to have_http_status(:ok) + expect(thread.reload.status).to eq("discarded") + end + + it "gives the plan author's own API threads the todo initial status" do + post api_v1_plan_comments_path(plan), params: { body_markdown: "note to self", agent_name: "Claude" }, headers: agent_headers, as: :json + expect(response).to have_http_status(:created) + expect(JSON.parse(response.body)["status"]).to eq("todo") + end + end + + def create_agent_collab_session(token) + CoPlan::AgentSession.create!( + plan_id: plan.id, + api_token_id: token.id, + agent_name: token.agent_name, + state: "complete" + ) + end +end diff --git a/spec/services/plans/replace_content_spec.rb b/spec/services/plans/replace_content_spec.rb index 9a462875..865cdcc8 100644 --- a/spec/services/plans/replace_content_spec.rb +++ b/spec/services/plans/replace_content_spec.rb @@ -31,7 +31,8 @@ let(:new_content) { initial_content.sub("unit tests", "integration tests") } it "broadcasts the new content body so other tabs live-update" do - expect(CoPlan::Broadcaster).to receive(:replace_plan_content).with(plan).and_call_original + expect(CoPlan::Broadcaster).to receive(:replace_plan_content) + .with(plan, changed_sections: kind_of(Array)).and_call_original allow(Turbo::StreamsChannel).to receive(:broadcast_stream_to) allow(Turbo::StreamsChannel).to receive(:broadcast_replace_to) From 124afd04d5c05272be13a43f95dbd35ef00ccbe0 Mon Sep 17 00:00:00 2001 From: Hampton Lintorn-Catlin Date: Fri, 7 Aug 2026 17:10:02 -0500 Subject: [PATCH 02/35] Voice tier, collaboration docs, and API orphan-thread fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Push-to-talk mic button on the plan page (Web Speech API tier): speaks feedback into a comment, listens for the agent pill to speak "Got it." / "Done — take a look." cues. Hidden when unsupported. - voice/ Pipecat sidecar scaffold (MLX Whisper + Kokoro via OpenAI-compatible endpoint) documenting the higher-fidelity local pipeline and its provider plugin seams. - docs/AGENT_COLLABORATION.md: the live feedback loop, bridge config, per-harness adapter recipes, permission-posture caveat, local demo. - API fix: wrap thread + first comment creation in a transaction so a failed comment (e.g. missing agent_name) can't leave an orphan empty thread; regression spec included. - Bridge: unbuffered stdout, drop a dead line. Co-Authored-By: Claude Fable 5 --- docs/AGENT_COLLABORATION.md | 85 +++++++++++ .../assets/stylesheets/coplan/application.css | 23 +++ .../coplan/api/v1/comments_controller.rb | 21 ++- .../controllers/coplan/voice_controller.js | 140 ++++++++++++++++++ .../app/views/coplan/plans/_toolbar.html.erb | 14 ++ script/coplan-bridge | 3 +- spec/requests/api/v1/agent_events_spec.rb | 9 ++ voice/README.md | 54 +++++++ voice/bot.py | 97 ++++++++++++ 9 files changed, 437 insertions(+), 9 deletions(-) create mode 100644 docs/AGENT_COLLABORATION.md create mode 100644 engine/app/javascript/controllers/coplan/voice_controller.js mode change 100644 => 100755 script/coplan-bridge create mode 100644 voice/README.md create mode 100644 voice/bot.py diff --git a/docs/AGENT_COLLABORATION.md b/docs/AGENT_COLLABORATION.md new file mode 100644 index 00000000..40b741e7 --- /dev/null +++ b/docs/AGENT_COLLABORATION.md @@ -0,0 +1,85 @@ +# Agent collaboration: the live feedback loop + +How an agent that authored a plan hears about your comment within a +second, shows you it's working, replies, and edits the doc while you +watch. Design docs (options + tradeoffs) live in CoPlan itself — see the +"Agent Collaboration in CoPlan" umbrella plan. + +## The loop + +1. A human comments (typed, or spoken via the mic button). +2. The comment fans out to the **agent event inbox** (`AgentEvent`) of + every agent session on the plan — never back to the actor itself. +3. The agent (or `script/coplan-bridge` on its behalf) is long-polling + `GET /api/v1/agent/events` and wakes. +4. It flips its **agent session** to `active` (the presence pill humans + see), replies on the thread narrating what it'll change, PUTs the + edit, and lands on `complete`. +5. Every open tab gets the new content over Turbo Streams; the sections + that changed **flash** — word-level ins/del for edited paragraphs — + then settle. + +The delivery is pull-based on purpose: agents run on laptops behind NAT, +so CoPlan never assumes it can push. Same endpoint serves long-poll +(plain JSON, curl-friendly) and SSE (`Accept: text/event-stream`). +Cursor = last event id (UUIDv7, time-ordered); delivery is +at-least-once with explicit ack. + +Full API reference: `GET /agent-instructions` → "Live Collaboration". + +## The bridge + +`script/coplan-bridge --config bridge.json` claims sessions on the plans +you list, drains the inbox, and injects each event into your harness via +a "resume session with message" command. It flips the pill to `active` +before the harness even boots, so the human sees life immediately. + +```json +{ + "base_url": "http://localhost:3222", + "token": "", + "agent_name": "Claude", + "plans": [""], + "adapter": "claude", + "harness_session": "" +} +``` + +### Per-harness adapters + +| `adapter` | Command shape (exec-resume, works everywhere) | Better, push-into-live-session surface | +|---|---|---| +| `claude` | `claude -p --resume ` | Channels MCP server (research preview) or Agent SDK streaming input | +| `codex` | `codex exec resume ` | `codex app-server` → `turn/start` / `turn/steer` | +| `goose` | `goose run --name --resume -t ` | `goose serve` (ACP) → `session/prompt` | +| `openhands` | `openhands --headless --resume -t ` | agent-server `POST /conversations/:id/events` | +| `amp` | `amp threads continue -x ` | `amp -x --stream-json-input` with `"steer": true` | +| `demo` | in-process deterministic agent (ack → reply → small edit) — no harness or tokens needed | — | + +Unattended runs need each harness's permission-relaxation flag +(`--permission-mode acceptEdits`, `--full-auto`, `GOOSE_MODE=auto`, …). +The default `claude` adapter uses `acceptEdits`; choose your own posture +deliberately — the bridge never escalates beyond what the config says. + +An agent doesn't need the bridge at all: any agent that can run curl in +a loop can follow the "Live Collaboration" section of +`/agent-instructions` directly. + +## Demoing locally + +```bash +bin/rails server -p 3222 +# Make a token (Settings → API Tokens) for the user who authors the plan. + +# Terminal 2 — the "agent": +script/coplan-bridge --config bridge.json # adapter: "demo" for zero-cost + +# Browser: open the plan, comment on a stiff sentence +# ("this is way too formal"). Watch: pill appears → reply lands → +# section flashes with word-level diffs. +``` + +Voice: the mic button on the plan page (Chrome) speaks your feedback into +the same loop, with spoken "Got it." / "Done — take a look." cues. The +higher-fidelity local pipeline (Pipecat + MLX Whisper + Kokoro) lives in +`voice/`. diff --git a/engine/app/assets/stylesheets/coplan/application.css b/engine/app/assets/stylesheets/coplan/application.css index e4c66398..dd627711 100644 --- a/engine/app/assets/stylesheets/coplan/application.css +++ b/engine/app/assets/stylesheets/coplan/application.css @@ -861,6 +861,29 @@ del.agent-flash { del.agent-flash { animation: none; display: none; } } +/* Voice push-to-talk — mic button + transient transcript/status text. */ +.voice-control { + display: inline-flex; + align-items: center; + gap: var(--space-sm); +} + +.voice-btn--listening { + color: var(--color-danger); + animation: agent-pill-pulse 1.2s ease-in-out infinite; +} + +.voice-status { + max-width: 16rem; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.voice-status--error { + color: var(--color-danger); +} + /* Flash messages */ .flash { padding: var(--space-md) var(--space-lg); diff --git a/engine/app/controllers/coplan/api/v1/comments_controller.rb b/engine/app/controllers/coplan/api/v1/comments_controller.rb index 89201a58..ed62ab26 100644 --- a/engine/app/controllers/coplan/api/v1/comments_controller.rb +++ b/engine/app/controllers/coplan/api/v1/comments_controller.rb @@ -34,14 +34,19 @@ def create status: initial_status ) - thread.save! - - comment = thread.comments.create!( - author_type: api_author_type, - author_id: current_user&.id, - body_markdown: params[:body_markdown], - agent_name: params[:agent_name] - ) + # Atomic, matching the web flow: a thread whose first comment + # fails validation (e.g. missing agent_name) must not survive as + # an empty orphan with a live anchor highlight. + comment = nil + ActiveRecord::Base.transaction do + thread.save! + comment = thread.comments.create!( + author_type: api_author_type, + author_id: current_user&.id, + body_markdown: params[:body_markdown], + agent_name: params[:agent_name] + ) + end reason = comment.agent? ? "agent_response" : "new_comment" CreateNotificationsJob.perform_later( diff --git a/engine/app/javascript/controllers/coplan/voice_controller.js b/engine/app/javascript/controllers/coplan/voice_controller.js new file mode 100644 index 00000000..74137434 --- /dev/null +++ b/engine/app/javascript/controllers/coplan/voice_controller.js @@ -0,0 +1,140 @@ +import { Controller } from "@hotwired/stimulus" + +/* + * coplan--voice + * + * Push-to-talk feedback for the plan's agent. Hold the mic button (or tap + * to toggle), say "this section is way too formal", and the transcript is + * posted as a comment — which wakes the plan's agent through the normal + * event inbox. The agent's session pill flips to active, and this + * controller speaks a short acknowledgment ("Got it.") the moment that + * happens, then a wrap-up cue when the session completes, so the loop is + * ear-and-eyes: you hear the ack, you watch the diff flashes land. + * + * This is the zero-install voice tier: browser-native SpeechRecognition + * (on-device in Chrome 139+) and speechSynthesis. The higher-fidelity + * OSS sidecar (Pipecat + local Whisper + Kokoro over WebRTC) plugs into + * the same comment-driven loop — see voice/README.md — so this controller + * is also its fallback. + */ +export default class extends Controller { + static targets = ["button", "status"] + static values = { url: String } + + connect() { + const Recognition = window.SpeechRecognition || window.webkitSpeechRecognition + if (!Recognition) { + this.element.style.display = "none" + return + } + + this.recognition = new Recognition() + this.recognition.continuous = false + this.recognition.interimResults = true + this.recognition.lang = document.documentElement.lang || "en-US" + this.recognition.onresult = (e) => this._onResult(e) + this.recognition.onend = () => this._onEnd() + this.recognition.onerror = (e) => this._setStatus(e.error === "no-speech" ? "Didn't catch that" : "Mic error", true) + + this.listening = false + this.finalTranscript = "" + this._watchAgentPill() + } + + disconnect() { + this.recognition?.abort() + this.pillObserver?.disconnect() + } + + toggle() { + this.listening ? this.recognition.stop() : this._start() + } + + _start() { + this.finalTranscript = "" + this.awaitingAck = false + this.listening = true + this.buttonTarget.classList.add("voice-btn--listening") + this._setStatus("Listening…") + this.recognition.start() + } + + _onResult(event) { + let interim = "" + for (const result of event.results) { + if (result.isFinal) this.finalTranscript += result[0].transcript + else interim += result[0].transcript + } + this._setStatus(`“${(this.finalTranscript + interim).trim().slice(-80)}”`) + } + + _onEnd() { + this.listening = false + this.buttonTarget.classList.remove("voice-btn--listening") + const text = this.finalTranscript.trim() + if (text.length === 0) return + + this._post(text) + } + + async _post(text) { + this._setStatus("Sending…") + const token = document.querySelector('meta[name="csrf-token"]')?.content + const body = new FormData() + body.append("comment_thread[body_markdown]", `🎙️ ${text}`) + + const response = await fetch(this.urlValue, { + method: "POST", + headers: { "X-CSRF-Token": token, "Accept": "text/vnd.turbo-stream.html, text/html" }, + body + }) + + if (response.ok) { + this.awaitingAck = true + this._setStatus("Sent — waiting for the agent…") + // If no agent picks it up shortly, stop promising. + setTimeout(() => { + if (this.awaitingAck) this._setStatus("") + this.awaitingAck = false + }, 20000) + } else { + this._setStatus("Couldn't send", true) + } + } + + // The agent session pill is broadcast-replaced whole; watch it and turn + // its state changes into speech. active → "Got it." (once per request), + // pill gone/complete → "Done — take a look." + _watchAgentPill() { + const pillHost = document.getElementById("plan-agent-sessions")?.parentNode + if (!pillHost) return + + this.pillObserver = new MutationObserver(() => { + const active = document.querySelector(".agent-pill--active, .agent-pill--pending") + if (this.awaitingAck && active) { + this.awaitingAck = false + this.spokeAck = true + this._speak("Got it.") + this._setStatus("Agent is on it…") + } else if (this.spokeAck && !active) { + this.spokeAck = false + this._speak("Done — take a look.") + this._setStatus("") + } + }) + this.pillObserver.observe(pillHost, { childList: true, subtree: true }) + } + + _speak(text) { + if (!window.speechSynthesis) return + const utterance = new SpeechSynthesisUtterance(text) + utterance.rate = 1.1 + window.speechSynthesis.speak(utterance) + } + + _setStatus(text, isError = false) { + if (!this.hasStatusTarget) return + this.statusTarget.textContent = text + this.statusTarget.classList.toggle("voice-status--error", isError) + } +} diff --git a/engine/app/views/coplan/plans/_toolbar.html.erb b/engine/app/views/coplan/plans/_toolbar.html.erb index f606a4b9..31c02c02 100644 --- a/engine/app/views/coplan/plans/_toolbar.html.erb +++ b/engine/app/views/coplan/plans/_toolbar.html.erb @@ -14,6 +14,20 @@ <% my_placement ||= plan.placements.includes(:library, :folder).detect { |p| p.library.writable_by?(current_user) } if current_user %> <% can_edit = allowed_to?(plan, :edit_content?) %> + <%# Server-side create errors land here (turbo_stream.update) — e.g. a + selection whose anchor doesn't resolve against the plan source. %> +
diff --git a/spec/factories/comment_threads.rb b/spec/factories/comment_threads.rb index 68a23b92..f486f183 100644 --- a/spec/factories/comment_threads.rb +++ b/spec/factories/comment_threads.rb @@ -6,8 +6,11 @@ status { "pending" } out_of_date { false } + # Threads refuse anchors that don't resolve against the plan content, + # so the anchor here is a real substring of the plan factory's default + # content_markdown. trait :with_anchor do - anchor_text { "some anchor text" } + anchor_text { "Some content here" } end trait :with_positioned_anchor do diff --git a/spec/models/comment_thread_anchor_spec.rb b/spec/models/comment_thread_anchor_spec.rb index 5d8c5b76..7dc910d8 100644 --- a/spec/models/comment_thread_anchor_spec.rb +++ b/spec/models/comment_thread_anchor_spec.rb @@ -13,6 +13,48 @@ plan end + describe "anchor_must_resolve on create" do + # A thread whose anchor never resolved renders nowhere: no highlight, + # no popover, no path to it from the page. Refused at the door rather + # than created invisible. + it "refuses a thread whose anchor resolves nowhere" do + expect { + plan.comment_threads.create!( + plan_version: plan.current_plan_version, + created_by_user: user, anchor_text: "text the plan never says" + ) + }.to raise_error(ActiveRecord::RecordInvalid, /nowhere to appear/) + end + + it "refuses an occurrence beyond the ones that exist" do + expect { + plan.comment_threads.create!( + plan_version: plan.current_plan_version, + created_by_user: user, anchor_text: "unit tests", anchor_occurrence: 3 + ) + }.to raise_error(ActiveRecord::RecordInvalid) + end + + it "allows a thread with no anchor at all" do + thread = plan.comment_threads.create!( + plan_version: plan.current_plan_version, created_by_user: user + ) + expect(thread).to be_persisted + end + + # Content drift after creation is the out_of_date flow, not a validity + # problem — an old thread must stay updatable. + it "does not re-litigate the anchor on update" do + thread = plan.comment_threads.create!( + plan_version: plan.current_plan_version, + created_by_user: user, anchor_text: "unit tests" + ) + thread.update_columns(anchor_text: "text no longer in the plan", anchor_start: nil, anchor_end: nil) + + expect(thread.reload.update(status: "todo")).to be true + end + end + describe "resolve_anchor_position on create" do it "resolves anchor_text to character positions" do thread = plan.comment_threads.create!( @@ -153,6 +195,13 @@ def assert_anchor_resolves(markdown, dom_text, expected_raw, occurrence: nil) assert_anchor_resolves(md, "run", "run") end + it "mermaid label text broken by
tags" do + md = "```mermaid\nflowchart LR\n Queue[\"assignment — first
fetching device wins\"] --> Printer\n```" + # The browser reads the rendered label back without the tag — + # "first
fetching" is selected as "firstfetching". + assert_anchor_resolves(md, "firstfetching device wins", "first
fetching device wins") + end + it "heading text (strips # markers)" do assert_anchor_resolves( "# My Heading\n\nContent here.", @@ -238,7 +287,7 @@ def assert_anchor_resolves(markdown, dom_text, expected_raw, occurrence: nil) version2 = CoPlan::PlanVersion.create!( plan: plan, revision: 2, content_markdown: new_content, actor_type: "human", actor_id: user.id, - operations_json: [{ "op" => "replace_exact", "resolved_range" => [anchor_pos, anchor_pos + 7], "new_range" => [anchor_pos, anchor_pos + 7], "delta" => 0 }] + operations_json: [ { "op" => "replace_exact", "resolved_range" => [ anchor_pos, anchor_pos + 7 ], "new_range" => [ anchor_pos, anchor_pos + 7 ], "delta" => 0 } ] ) plan.update!(current_plan_version: version2, current_revision: 2) @@ -259,7 +308,7 @@ def assert_anchor_resolves(markdown, dom_text, expected_raw, occurrence: nil) version2 = CoPlan::PlanVersion.create!( plan: plan, revision: 2, content_markdown: new_content, actor_type: "human", actor_id: user.id, - operations_json: [{ "op" => "replace_exact", "resolved_range" => [unit_test_pos, unit_test_pos + 10], "new_range" => [unit_test_pos, unit_test_pos + 17], "delta" => 7 }] + operations_json: [ { "op" => "replace_exact", "resolved_range" => [ unit_test_pos, unit_test_pos + 10 ], "new_range" => [ unit_test_pos, unit_test_pos + 17 ], "delta" => 7 } ] ) plan.update!(current_plan_version: version2, current_revision: 2) @@ -283,7 +332,7 @@ def assert_anchor_resolves(markdown, dom_text, expected_raw, occurrence: nil) version2 = CoPlan::PlanVersion.create!( plan: plan, revision: 2, content_markdown: new_content, actor_type: "human", actor_id: user.id, - operations_json: [{ "op" => "replace_exact", "resolved_range" => [first_pos, first_pos + first_len], "new_range" => [first_pos, first_pos + new_len], "delta" => new_len - first_len }] + operations_json: [ { "op" => "replace_exact", "resolved_range" => [ first_pos, first_pos + first_len ], "new_range" => [ first_pos, first_pos + new_len ], "delta" => new_len - first_len } ] ) plan.update!(current_plan_version: version2, current_revision: 2) @@ -305,7 +354,7 @@ def assert_anchor_resolves(markdown, dom_text, expected_raw, occurrence: nil) version2 = CoPlan::PlanVersion.create!( plan: plan, revision: 2, content_markdown: new_content, actor_type: "human", actor_id: user.id, - operations_json: [{ "op" => "replace_exact", "resolved_range" => [anchor_pos, anchor_pos + 7], "new_range" => [anchor_pos, anchor_pos + 7], "delta" => 0 }] + operations_json: [ { "op" => "replace_exact", "resolved_range" => [ anchor_pos, anchor_pos + 7 ], "new_range" => [ anchor_pos, anchor_pos + 7 ], "delta" => 0 } ] ) plan.update!(current_plan_version: version2, current_revision: 2) @@ -317,12 +366,12 @@ def assert_anchor_resolves(markdown, dom_text, expected_raw, occurrence: nil) describe "#anchor_valid?" do it "returns true for non-outdated thread" do - thread = create(:comment_thread, plan: plan, anchor_text: "some text") + thread = create(:comment_thread, plan: plan, anchor_text: "First section") expect(thread.anchor_valid?).to be true end it "returns false for outdated thread" do - thread = create(:comment_thread, plan: plan, anchor_text: "some text", out_of_date: true) + thread = create(:comment_thread, plan: plan, anchor_text: "First section", out_of_date: true) expect(thread.anchor_valid?).to be false end diff --git a/spec/requests/api/v1/plans_spec.rb b/spec/requests/api/v1/plans_spec.rb index 39c64c59..2ea5c216 100644 --- a/spec/requests/api/v1/plans_spec.rb +++ b/spec/requests/api/v1/plans_spec.rb @@ -263,13 +263,13 @@ def alice_placement it "comments returns thread list with anchor_text" do thread = create(:comment_thread, :with_anchor, plan: plan, - plan_version: plan.current_plan_version, created_by_user: alice, anchor_text: "original roadmap text") + plan_version: plan.current_plan_version, created_by_user: alice) get comments_api_v1_plan_path(plan), headers: headers expect(response).to have_http_status(:success) threads = JSON.parse(response.body) expect(threads).to be_a(Array) matching = threads.find { |t| t["id"] == thread.id } - expect(matching["anchor_text"]).to eq("original roadmap text") + expect(matching["anchor_text"]).to eq("Some content here") end describe "GET /api/v1/plans/:id/snapshot" do diff --git a/spec/requests/comment_threads_spec.rb b/spec/requests/comment_threads_spec.rb index 8b0fa1c2..ed338bd6 100644 --- a/spec/requests/comment_threads_spec.rb +++ b/spec/requests/comment_threads_spec.rb @@ -3,7 +3,16 @@ RSpec.describe "CommentThreads", type: :request do let(:alice) { create(:coplan_user, :admin) } let(:bob) { create(:coplan_user) } - let(:plan) { create(:plan, :considering, created_by_user: alice) } + + # Threads refuse anchors that don't resolve, so the plan has to actually + # say the thing these specs anchor to. + let(:plan) do + create(:plan, :considering, created_by_user: alice).tap do |p| + version = create(:plan_version, plan: p, revision: 2, actor_id: alice.id, + content_markdown: "## Ambition\n\nOur goal is world domination by Q3.\n") + p.update_columns(current_plan_version_id: version.id, current_revision: 2) + end + end before { sign_in_as(alice) } @@ -19,10 +28,37 @@ expect(response).to redirect_to(plan_path(plan)) thread = CoPlan::CommentThread.last expect(thread.anchor_text).to eq("world domination") + expect(thread.anchor_start).to be_present # resolved at the door expect(thread.status).to eq("todo") # author's own comments start as todo expect(thread.plan_version_id).to eq(plan.current_plan_version_id) end + # A thread whose anchor never resolved renders nowhere — no highlight, no + # popover, no way to reach it. "Comment posted" followed by nothing + # visible is worse than a refusal. + describe "when the anchor doesn't resolve against the plan" do + it "refuses to create the thread" do + expect { + post plan_comment_threads_path(plan), params: { + comment_thread: { anchor_text: "text the plan never says", body_markdown: "Lost forever." } + } + }.not_to change { [ CoPlan::CommentThread.count, CoPlan::Comment.count ] } + + expect(response).to redirect_to(plan_path(plan)) + expect(flash[:alert]).to include("nowhere to appear") + end + + it "tells a turbo-stream client with a 422 so it can fall back" do + post plan_comment_threads_path(plan), + params: { comment_thread: { anchor_text: "text the plan never says", body_markdown: "Lost." } }, + headers: { "Accept" => "text/vnd.turbo-stream.html, text/html" } + + expect(response).to have_http_status(:unprocessable_content) + expect(response.body).to include("new-comment-form-error") + expect(response.body).to include("nowhere to appear") + end + end + it "broadcasts the popover via requestless partial render, never request-scoped HTML" do # The popover contains reply/action forms; request-rendered HTML embeds # the actor's session authenticity token, which must not be broadcast. diff --git a/spec/services/coplan/comments/interpret_dictation_spec.rb b/spec/services/coplan/comments/interpret_dictation_spec.rb index 53c7f8e1..486c1f86 100644 --- a/spec/services/coplan/comments/interpret_dictation_spec.rb +++ b/spec/services/coplan/comments/interpret_dictation_spec.rb @@ -138,11 +138,13 @@ def stub_ai(text:, span: nil) expect(document).to include(result.anchor_text) end - # Inline markup breaks single-line spans the same way: the model reads - # "main is always releasable" off the screen, and the markdown says - # "`main` is always releasable". This one shipped as a comment pinned - # to the page heading instead of the word it asked about. - it "narrows a span broken by inline markup to the words that resolve" do + # Inline markup is different: the model reads "main is always + # releasable" off the screen, the markdown says "`main` is always + # releasable" — but CommentThread resolves exactly this via stripped + # markdown, so the whole span survives. (An earlier version narrowed + # it to "is always releasable": stricter than the resolver, and the + # highlight missed the very word the remark was about.) + it "keeps a span broken only by inline markup — the resolver handles it" do stub_ai(text: "What does releasable mean?", span: "main is always releasable") result = described_class.call( @@ -151,7 +153,7 @@ def stub_ai(text:, span: nil) transcript: "what does releasable mean" ) - expect(result.anchor_text).to eq("is always releasable") + expect(result.anchor_text).to eq("main is always releasable") end it "does not salvage a fragment too short to point at anything" do diff --git a/spec/system/mermaid_anchor_spec.rb b/spec/system/mermaid_anchor_spec.rb index 3f1487a1..6da3ee2f 100644 --- a/spec/system/mermaid_anchor_spec.rb +++ b/spec/system/mermaid_anchor_spec.rb @@ -57,8 +57,11 @@ def wait_for_diagram # An anchor that carries stylesheet text — the shape of anchors captured # by sweeping a selection across a diagram before capture excluded # non-rendered text. This string appears verbatim in the