Sign in

flagon-io/g1t

public

g1t/services/work/src/messages.rs

154 lines5,316 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Usage, like a hosting provider's: what agents cost, per day, task, repository and pull request1//! Steering: people send the agent at work on a pull request a message,
2//! and the agent receives it at its next step. The sandbox asks for
3//! undelivered messages after each of the agent's tool calls.
4
5use g1t_contracts::time::rfc3339;
6use g1t_contracts::work::*;
7use g1t_contracts::{FailureCode, Outcome, PrincipalKind, new_id};
8use g1t_kit::now_ms;
9use serde::Deserialize;
10use worker::Result;
11
12use crate::Work;
13
14/// The longest message an agent is sent.
15const MAX_MESSAGE_CHARS: usize = 4000;
16
17#[derive(Deserialize)]
18struct MessageRow {
19 id: String,
20 author_name: String,
21 body: String,
22 created_at: String,
23 delivered_at: Option<String>,
24}
25
26impl From<MessageRow> for AgentMessage {
27 fn from(row: MessageRow) -> Self {
28 AgentMessage {
29 id: row.id,
30 author: row.author_name,
31 body: row.body,
32 created_at: row.created_at,
33 delivered_at: row.delivered_at,
34 }
35 }
36}
37
38impl Work {
39 /// Every message sent to the agent on a pull request, oldest first.
40 pub(crate) async fn messages(&self, pull_id: &str) -> Result<Vec<AgentMessage>> {
41 Ok(self
42 .db
43 .prepare("SELECT * FROM agent_messages WHERE pull_id = ? ORDER BY created_at, id")
44 .bind(&[pull_id.into()])?
45 .all()
46 .await?
47 .results::<MessageRow>()?
48 .into_iter()
49 .map(AgentMessage::from)
50 .collect())
51 }
52
53 pub(crate) async fn message_agent(&self, a: MessageAgentArgs) -> Result<Outcome<AgentMessage>> {
54 let viewer = Some(a.actor.clone());
55 let (repo, pull) = match self.pull_at(&a.repo, a.number, &viewer).await? {
56 Outcome::Ok(found) => found,
57 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
58 };
59 if !a.actor.verified
60 || (pull.author.id != a.actor.id && !a.actor.is_member(&repo.namespace))
61 {
62 return Ok(Outcome::fail(
63 FailureCode::Forbidden,
64 "Only the pull request's author and members of the workspace can message its agent.",
65 ));
66 }
67 if !pull.status.is_active() {
68 return Ok(Outcome::fail(
69 FailureCode::Conflict,
70 "This pull request is no longer being worked on.",
71 ));
72 }
73 let body = a.body.trim();
74 if body.is_empty() {
75 return Ok(Outcome::fail(FailureCode::Invalid, "Write a message."));
76 }
77 let body: String = body.chars().take(MAX_MESSAGE_CHARS).collect();
78 let now = now_ms();
79 let message = AgentMessage {
80 id: new_id("msg", now),
81 author: a.actor.username.clone(),
82 body,
83 created_at: rfc3339(now),
84 delivered_at: None,
85 };
86 self.db
87 .prepare(
88 "INSERT INTO agent_messages (id, pull_id, author_id, author_name, body, created_at)
89 VALUES (?, ?, ?, ?, ?, ?)",
90 )
91 .bind(&[
92 message.id.as_str().into(),
93 pull.id.as_str().into(),
94 a.actor.id.as_str().into(),
95 message.author.as_str().into(),
96 message.body.as_str().into(),
97 message.created_at.as_str().into(),
98 ])?
99 .run()
100 .await?;
101 self.note(
102 &repo.id,
103 pull.number,
104 (a.actor.id.as_str(), a.actor.username.as_str()),
105 "sent the agent a message",
106 )
107 .await?;
108 Ok(Outcome::Ok(message))
109 }
110
111 /// The undelivered messages, marked delivered and recorded in the
112 /// session, for the agent's sandbox.
113 pub(crate) async fn take_messages(&self, a: TakeMessagesArgs) -> Result<Outcome<Vec<AgentMessage>>> {
114 if a.actor.kind != PrincipalKind::Agent {
115 return Ok(Outcome::fail(
116 FailureCode::Forbidden,
117 "Only g1t's agents take messages.",
118 ));
119 }
120 let viewer = Some(a.actor.clone());
121 let (_, pull) = match self.pull_at(&a.repo, a.number, &viewer).await? {
122 Outcome::Ok(found) => found,
123 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
124 };
125 let now = rfc3339(now_ms());
126 let taken: Vec<AgentMessage> = self
127 .db
128 .prepare(
129 "UPDATE agent_messages SET delivered_at = ?
130 WHERE pull_id = ? AND delivered_at IS NULL
131 RETURNING *",
132 )
133 .bind(&[now.as_str().into(), pull.id.as_str().into()])?
134 .all()
135 .await?
136 .results::<MessageRow>()?
137 .into_iter()
138 .map(AgentMessage::from)
139 .collect();
140 if !taken.is_empty() {
141 let entries: Vec<NewSessionEntry> = taken
142 .iter()
143 .map(|message| NewSessionEntry {
144 kind: SessionEntryKind::Prompt,
145 text: format!("Message from {}: {}", message.author, message.body),
146 tool: None,
147 commit: None,
148 })
149 .collect();
150 self.append_entries(&pull, &entries).await?;
151 }
152 Ok(Outcome::Ok(taken))
153 }
154}