Drafts stream into the editor as the assistant writes them 119fd17d213f

/* Drafting with an assistant (oxygenhub#248): the draft is streamed into

   the description as it is written, so a long one is neither cut off by a

   proxy nor waited for in silence. Without this file the button posts the

   form and the page comes back with the draft. */

(function () {

  "use strict";

  if (!window.fetch || !window.TextDecoder || !window.URLSearchParams) return;

  document.querySelectorAll("button.agent-btn[data-stream]").forEach(function (btn) {

    btn.addEventListener("click", function (e) {

      var form = btn.form, area = form && form.querySelector('textarea[name="body"]');

      if (!form || !area) return;

      e.preventDefault();

      function get() { return area._mde ? area._mde.value() : area.value; }

      function show(t) { if (area._mde) area._mde.value(t); else area.value = t; }

      var before = get(), text = "";

      var box = (area._mde && area.parentNode.querySelector(".EasyMDEContainer")) || area;

      var note = form.querySelector(".draft-note");

      if (!note) { note = document.createElement("p"); note.className = "draft-note small mute"; box.parentNode.insertBefore(note, box.nextSibling); }

      note.textContent = btn.value + " is drafting the description…";

      var buttons = form.querySelectorAll("button.agent-btn");

      buttons.forEach(function (b) { b.disabled = true; });

      var data = new URLSearchParams(new FormData(form));

      data.set("assistant", btn.value);

      var finished = false;

      function end(message, keep) { finished = true; if (!keep) show(before); note.textContent = message; }

      fetch(btn.dataset.stream, { method: "POST", body: data, credentials: "same-origin" })

        .then(function (r) {

          if (!r.ok || !r.body) throw new Error("the hub answered " + r.status);

          var reader = r.body.getReader(), dec = new TextDecoder(), buf = "";

          function pump() {

            return reader.read().then(function (res) {

              if (res.done) { if (!finished) end(btn.value + " stopped before finishing.", text !== ""); return; }

              buf += dec.decode(res.value, { stream: true });

              var i;

              while ((i = buf.indexOf("\n\n")) >= 0) {

                var block = buf.slice(0, i), kind = "message", payload = "";

                buf = buf.slice(i + 2);

                block.split("\n").forEach(function (l) {

                  if (l.indexOf("event:") === 0) kind = l.slice(6).trim();

                  else if (l.indexOf("data:") === 0) payload += l.slice(5).trim();

                });

                if (!payload) continue;

                var v;

                try { v = JSON.parse(payload); } catch (_) { continue; }

                if (kind === "part") { text += v; show(text); }

                else if (kind === "done") { show(v); end("Drafted by " + btn.value + "; nothing is saved until you save the proposal.", true); return; }

                else if (kind === "error") { end(v, false); return; }

              }

              return pump();

            });

          }

          return pump();

        })

        .catch(function (err) { end(btn.value + " did not answer: " + err.message + ".", false); })

        .then(function () { buttons.forEach(function (b) { b.disabled = false; }); });

    });

  });

})();

    ask_within(app, a, &system, budget, BACKGROUND_TIMEOUT, |room| review_context(row, p, &loaded, room), |_| {}).await

    mut part: impl FnMut(&str),

    // A refusal for length comes before any part, so asking again starts

    // the answer afresh.

    match ask_streaming(app, a, system, &context(room(budget)), wait, &mut part).await {

            ask_streaming(app, a, system, &context(room(smaller)), wait, &mut part).await.map_err(|e2| {

    ask_within(app, a, &system, budget, BACKGROUND_TIMEOUT, |room| review_context(row, p, &loaded, room), |_| {}).await

    app: &Shared,

    row: &crate::db::repos::RepoRow,

    by: &crate::db::accounts::Account,

    a: &Assistant,

    d: Draft<'_>,

) -> Result<String, String> {

    draft_description_streaming(app, row, by, a, d, |_| {}).await

}

/// [`draft_description`], handing each part of the draft to `part` as it

/// comes, for a page that shows it being written (oxygenhub#248).

pub async fn draft_description_streaming(

    part: impl FnMut(&str),

            ask_within(app, a, &system, budget, CALL_TIMEOUT, |room| draft_context(row, title, target, changes, diffs, room), part).await

        .route("/{owner}/{name}/proposals/new/draft/stream", post(proposals::new_draft_stream))

        .route("/{owner}/{name}/proposal/{number}/draft/stream", post(proposals::draft_stream))

/// A draft being written, as server-sent events for the page: `part` with

/// each piece of text, then `done` with the whole draft or `error` with

/// why there is none (each a JSON string). The response starts at once, so

/// a draft that takes minutes is not cut off by a proxy waiting for it

/// (oxygenhub#248).

type DraftEvents = axum::response::Sse<

    axum::response::sse::KeepAliveStream<

        tokio_stream::wrappers::ReceiverStream<Result<axum::response::sse::Event, std::convert::Infallible>>,

    >,

>;

fn draft_events(rx: tokio::sync::mpsc::Receiver<Result<axum::response::sse::Event, std::convert::Infallible>>) -> DraftEvents {

    axum::response::Sse::new(tokio_stream::wrappers::ReceiverStream::new(rx))

        .keep_alive(axum::response::sse::KeepAlive::new().interval(std::time::Duration::from_secs(15)))

}

fn draft_event(kind: &str, text: &str) -> Result<axum::response::sse::Event, std::convert::Infallible> {

    Ok(axum::response::sse::Event::default().event(kind).data(serde_json::to_string(text).unwrap_or_default()))

}

/// Only an error, for a draft that cannot start.

fn draft_refused(why: String) -> DraftEvents {

    let (tx, rx) = tokio::sync::mpsc::channel(1);

    let _ = tx.try_send(draft_event("error", &why));

    draft_events(rx)

}

/// What a draft is written from, owned so a task of its own can have it.

struct DraftJob {

    row: db::repos::RepoRow,

    by: accounts::Account,

    a: crate::assistants::Assistant,

    slot: i64,

    title: String,

    target: String,

    changes: Vec<ChangeView>,

    diffs: Vec<oxy_core::repo::FileDiff>,

}

/// Write the draft in a task of its own, its parts to the page.

fn draft_in_background(app: Shared, job: DraftJob) -> DraftEvents {

    let (tx, rx) = tokio::sync::mpsc::channel(64);

    tokio::spawn(async move {

        let DraftJob { row, by, a, slot, title, target, changes, diffs } = job;

        let d = crate::assistants::Draft { slot, title: &title, target: &target, changes: &changes, diffs: &diffs };

        let parts = tx.clone();

        let result = crate::assistants::draft_description_streaming(&app, &row, &by, &a, d, move |p: &str| {

            // A page that went away stops nothing: the draft finishes.

            let _ = parts.try_send(draft_event("part", p));

        })

        .await;

        let last = match result {

            Ok(text) => draft_event("done", text.trim()),

            Err(why) => draft_event("error", &format!("{} did not answer: {why}.", a.name)),

        };

        let _ = tx.send(last).await;

    });

    draft_events(rx)

}

/// `POST /{owner}/{name}/proposal/{number}/draft/stream`: [`draft`], with

/// the draft streamed to the page as it is written.

pub async fn draft_stream(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, number)): Path<(String, String, i64)>,

    Form(f): Form<DraftForm>,

) -> Result<DraftEvents, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/proposal/{number}"), &viewer);

    let me = require(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let (repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    super::writable(&row, &shell)?;

    let p = pdb::get(&app.db, row.id, number)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .ok_or_else(|| Error::not_found(shell.clone()))?;

    if Some(me.account.id) != p.author_id && role < Role::Maintain {

        return Err(Error::not_found(shell));

    }

    let Some(a) = crate::assistants::enabled_for_repo(&app.db, &row)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?

        .into_iter()

        .find(|a| a.name.eq_ignore_ascii_case(f.assistant.trim()))

    else {

        return Ok(draft_refused(format!("No assistant called {} here.", f.assistant.trim())));

    };

    let loaded = load(&app, &repo, p.clone()).await.map_err(|e| Error::from_repo(shell.clone(), e))?;

    let title = if f.title.trim().is_empty() { p.title.clone() } else { f.title.clone() };

    let job = DraftJob {

        row,

        by: me.account.clone(),

        a,

        slot: p.id,

        title,

        target: p.target.clone(),

        changes: loaded.changes,

        diffs: loaded.diffs,

    };

    Ok(draft_in_background(app.clone(), job))

}

/// `POST /{owner}/{name}/proposals/new/draft/stream`: [`new_draft`], with

/// the draft streamed to the page as it is written.

pub async fn new_draft_stream(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name)): Path<(String, String)>,

    Form(f): Form<NewForm>,

) -> Result<DraftEvents, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/proposals/new"), &viewer);

    let me = require(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let (repo, row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    super::writable(&row, &shell)?;

    let Some(a) = crate::assistants::enabled_for_repo(&app.db, &row)

        .await

        .unwrap_or_default()

        .into_iter()

        .find(|a| a.name.eq_ignore_ascii_case(f.assistant.trim()))

    else {

        return Ok(draft_refused(format!("No assistant called {} here.", f.assistant.trim())));

    };

    let (changes, diffs) = match preview_diff(&app, &repo, &f.target, &f.source).await {

        Ok(x) => x,

        Err(e) => return Ok(draft_refused(e.to_string())),

    };

    if changes.is_empty() {

        return Ok(draft_refused("Nothing to describe: the source adds nothing to the target.".into()));

    }

    let job = DraftJob { slot: -row.id, row, by: me.account.clone(), a, title: f.title.clone(), target: f.target.clone(), changes, diffs };

    Ok(draft_in_background(app.clone(), job))

}

    // The same as a stream for the page to show as it comes (#248): the

    // button says where, and the draft arrives in parts, then whole.

    let page = get(app, "/andy/demo/proposal/1", Some(&bob)).await.body;

    assert!(page.contains("data-stream=\"/andy/demo/proposal/1/draft/stream\""), "{page}");

    let form = format!("_csrf={}&assistant=Claude&title=Add+three", csrf_of(&page));

    let r = post(app, "/andy/demo/proposal/1/draft/stream", Some(&bob), &form).await;

    assert_eq!(r.status, StatusCode::OK, "{}", r.body);

    assert!(r.body.contains("event: part") && r.body.contains("event: done"), "{}", r.body);

    assert!(r.body.contains("Adds a third line."), "{}", r.body);

    assert_eq!(db::proposals::get(&app.db, 1, 1).await.unwrap().unwrap().body, "Please", "still nothing saved");

    let r = post(app, "/andy/demo/proposal/1/draft/stream", Some(&bob), &form.replace("Claude", "Nobody")).await;

    assert!(r.body.contains("event: error") && r.body.contains("No assistant called Nobody"), "{}", r.body);

    seen.lock().unwrap().clear();

- **Draft the description.** The new-proposal form (anyone who may open one) and an open proposal's editor (the author or a maintainer) have a **draft with NAME** button per assistant. It asks the assistant, now, for a description from the title so far, the changes (titles and the first lines of their messages) and the diff, under a fixed system prompt (description field only, what and why, what to look at, how to verify) plus the levels' prompts, and puts the answer in the editor with a note; nothing is saved until the person updates or opens the proposal. The audit log records `assistant.drafted`; the same per-repository hourly limit applies. With JavaScript the button posts to `…/draft/stream` instead, which answers at once with server-sent events (`part` with each piece of text as the assistant writes it, then `done` with the whole draft or `error` with why there is none, each a JSON string, and a keep-alive every 15 s), and the editor fills as the draft comes; without it the form posts and the page returns with the draft. Every assistant call is made as a stream, and its time limit is for silence (90 s for a draft, 300 s for a review) under a 15-minute cap, so a long answer that keeps coming is not cut off (oxygenhub#248).

        <div class="tl-actions"><a class="go" href="#description">Cancel</a> {% for a in assistants %}<button type="submit" class="agent-btn" formaction="{{ repo.url }}/proposal/{{ p.number }}/draft" data-stream="{{ repo.url }}/proposal/{{ p.number }}/draft/stream" formnovalidate name="assistant" value="{{ a }}" title="{{ a }}, an AI assistant, drafts the description from the changes; you edit it before saving"><img src="/avatar/agent/{{ a }}" alt="" width="16" height="16"> Draft with {{ a }}</button> {% endfor %}<button type="submit" class="primary-btn">Update proposal</button></div>

  {% if !assistants.is_empty() %}<div class="side-assist">{% for a in assistants %}<button type="submit" class="agent-btn" formaction="{{ repo.url }}/proposals/new/draft" data-stream="{{ repo.url }}/proposals/new/draft/stream" formnovalidate name="assistant" value="{{ a }}" title="{{ a }}, an AI assistant, drafts the description from the changes; you edit it before opening"><img src="/avatar/agent/{{ a }}" alt="" width="16" height="16"> Draft with {{ a }}</button>{% endfor %}</div>{% endif %}