fix(claw): an account-watched thread resolves its client without a repo watch

`watch_forge` looked the dispatch key up in the `watches` table and
nothing else, so a thread that arrived through an *account* watch — the
daemon's own account notifications, which cover the whole instance, not
one repo — failed with "no watch configured" unless the same repo also
had a repo watch row. The dispatch is created by the account poll; the
client it needs is the account's. Nothing could hand it over, because the
payload did not name the account either.

The triaged payload now carries `account_id`, the dispatch drains the
inbox *before* resolving the client (the inbox is where that id lands),
and `watch_forge_as` falls back to the account watch that delivered the
news: its host, its credential, its login. A repo watch row still wins
when there is one — a thread with its own watch keeps its own identity —
and `watch_forge` keeps its signature for callers that know nothing about
accounts, falling back to the kind's first account watch.

The client cache is now one helper over `(forge, host, credential)`, so
the dispatch's replies ride the same pool the poll built.
This commit is contained in:
kosmos authored and Valentin Haudiquet committed 2026-11-08 17:54:15 +01:00
1 parent 838f71738e
commit 875b2f0ed8
2 files changed
+201 -23

No files matched your search

+5
View File
@@ -502,6 +502,11 @@ impl ForgeSource {
"is_pr": notification.is_pr, "is_pr": notification.is_pr,
"role": role.as_str(), "role": role.as_str(),
"notification": notification, "notification": notification,
// The account whose feed delivered this: the
// dispatch answers as that identity, and a thread
// with no repo watch row of its own has only this
// to resolve a client by.
"account_id": watch.account_id,
}); });
self.dispatcher self.dispatcher
.ingest(Event { .ingest(Event {
+196 -23
View File
@@ -23,7 +23,7 @@ use std::collections::HashMap;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, PoisonError}; use std::sync::{Arc, Mutex, PoisonError};
use crate::forge::{record_client, resolve_watch_credential}; use crate::forge::{WatchCredential, record_client, resolve_watch_credential};
use crate::watch_store::ClawStore; use crate::watch_store::ClawStore;
use rhema_agent_tools::AllowAll; use rhema_agent_tools::AllowAll;
use rhema_autonomous_agent::{Handler, SessionContext, async_trait}; use rhema_autonomous_agent::{Handler, SessionContext, async_trait};
@@ -97,46 +97,117 @@ impl ClawHandler {
/// The forge client a dispatch key answers to: /// The forge client a dispatch key answers to:
/// `"github:issue:acme/widgets#7"` resolves its watch record from the /// `"github:issue:acme/widgets#7"` resolves its watch record from the
/// table at dispatch time — the /// table at dispatch time — the table is the source of truth, not a
/// table is the source of truth, not a boot-time snapshot — and the /// boot-time snapshot — and the client is cached per
/// client is cached per `(forge, host, token)` so one poll pool serves /// `(forge, host, credential)` so one poll pool serves every thread of
/// every thread of a watch. /// a watch.
///
/// A thread an account watch delivered may have no repo watch of its
/// own: the account's notification feed covers the whole instance, so
/// the row that fed the dispatch is the account watch, and the reply
/// has to post as that account. `as_account` names it; `None` — the
/// caller does not know, or there is one account per forge — takes the
/// kind's first account watch.
pub fn watch_forge(&self, key: &ResourceKey) -> Result<Arc<dyn Forge>, String> { pub fn watch_forge(&self, key: &ResourceKey) -> Result<Arc<dyn Forge>, String> {
self.watch_forge_as(key, None)
}
/// [`Self::watch_forge`] with the account the dispatch was delivered
/// as, when the caller knows it: an account notification carries the
/// id it arrived on (see [`crate::forge::ForgeSource`]'s poll).
pub fn watch_forge_as(
&self,
key: &ResourceKey,
as_account: Option<&str>,
) -> Result<Arc<dyn Forge>, String> {
let watch = watch_key(key) let watch = watch_key(key)
.ok_or_else(|| format!("resource key {key} is not a forge thread key"))?; .ok_or_else(|| format!("resource key {key} is not a forge thread key"))?;
let (kind, repo) = watch let (forge, repo) = watch
.split_once(':') .split_once(':')
.ok_or_else(|| format!("watch entry {watch} has no forge kind"))?; .ok_or_else(|| format!("watch entry {watch} has no forge kind"))?;
let record = self let records = self
.watches .watches
.list() .list()
.map_err(|err| format!("watch lookup for {key} failed: {err}"))? .map_err(|err| format!("watch lookup for {key} failed: {err}"))?;
.into_iter() if let Some(record) = records
.find(|record| record.forge == kind && record.repo == repo) .iter()
.ok_or_else(|| format!("no watch configured for {key}"))?; .find(|record| record.forge == forge && record.repo == repo)
{
let kind = ForgeKind::parse(&record.forge) let kind = ForgeKind::parse(&record.forge)
.ok_or_else(|| format!("watch entry {watch} names an unknown forge"))?; .ok_or_else(|| format!("watch entry {watch} names an unknown forge"))?;
// The same chain the poll resolved: row token, env var, then // The same chain the poll resolved: row token, env var, then
// the matching account identity — keyed on the resolved value, // the matching account identity — keyed on the resolved
// so a credential rotation (or a saved account) is a new client // value, so a credential rotation (or a saved account) is a
// on the next dispatch. The login rides along: an instance that // new client on the next dispatch. The login rides along: an
// refuses the value in the token header takes it as Basic auth // instance that refuses the value in the token header takes it
// (`GiteaForge::with_login`). // as Basic auth (`GiteaForge::with_login`).
let credential = resolve_watch_credential( let credential = resolve_watch_credential(
self.accounts.as_ref(), self.accounts.as_ref(),
kind, kind,
record.host.as_deref(), record.host.as_deref(),
&record.token, &record.token,
); );
return Ok(self.client(kind, record.host.as_deref(), &credential));
}
// No repo watch: the account watch is what fed this thread, so its
// identity is what the reply posts as.
let account_watches = self
.watches
.list_account_watches()
.map_err(|err| format!("watch lookup for {key} failed: {err}"))?;
let account_watch = account_watches
.iter()
.filter(|watch| watch.forge == forge)
.find(|watch| as_account.is_none_or(|id| id == watch.account_id))
.ok_or_else(|| format!("no watch configured for {key}"))?;
let kind = ForgeKind::parse(&account_watch.forge)
.ok_or_else(|| format!("watch entry {watch} names an unknown forge"))?;
let secret = self
.accounts
.secret(&account_watch.account_id)
.map_err(|err| format!("account lookup for {key} failed: {err}"))?
.ok_or_else(|| {
format!(
"the account watch for {key} names account {}, which is gone",
account_watch.account_id
)
})?;
let login = self
.accounts
.get(&account_watch.account_id)
.map_err(|err| format!("account lookup for {key} failed: {err}"))?
.and_then(|account| account.username);
let credential = WatchCredential {
token: secret.value,
login,
};
Ok(self.client(kind, account_watch.host.as_deref(), &credential))
}
/// The cached client for one credential: the dispatch path and the
/// poll path build the same key, so a thread's replies ride the poll's
/// pool.
fn client(
&self,
kind: ForgeKind,
host: Option<&str>,
credential: &WatchCredential,
) -> Arc<dyn Forge> {
// Everything that changes the client: the kind, its host, and the
// credential — the login included, since one value sent two ways
// is two credentials.
let cache_key = format!( let cache_key = format!(
"{}|{:?}|{}|{:?}", "{}|{:?}|{}|{:?}",
record.forge, record.host, credential.token, credential.login kind.name(),
host,
credential.token,
credential.login
); );
let mut clients = self.clients.lock().unwrap_or_else(PoisonError::into_inner); let mut clients = self.clients.lock().unwrap_or_else(PoisonError::into_inner);
Ok(clients clients
.entry(cache_key) .entry(cache_key)
.or_insert_with(|| record_client(kind, record.host.as_deref(), &credential)) .or_insert_with(|| record_client(kind, host, credential))
.clone()) .clone()
} }
/// One dispatch, failed with a reason on any step. The session stash /// One dispatch, failed with a reason on any step. The session stash
@@ -148,13 +219,17 @@ impl ClawHandler {
let key = ctx.key.as_str().to_owned(); let key = ctx.key.as_str().to_owned();
let (repo, thread_kind, number) = parse_key(&ctx.key) let (repo, thread_kind, number) = parse_key(&ctx.key)
.ok_or_else(|| format!("resource key {key} is not a forge thread key"))?; .ok_or_else(|| format!("resource key {key} is not a forge thread key"))?;
let forge = self.watch_forge(&ctx.key)?; // Drained before the client resolves: an account notification
// names the account that delivered it, and the reply must post as
// that account — a thread the account's feed delivered may have no
// repo watch row at all.
let news = self.drain_news(ctx, &key).await?;
let forge = self.watch_forge_as(&ctx.key, delivery_account(&news))?;
let snapshot = forge let snapshot = forge
.snapshot(&repo, thread_kind, number) .snapshot(&repo, thread_kind, number)
.await .await
.map_err(|err| format!("snapshot of {key} failed: {err}"))?; .map_err(|err| format!("snapshot of {key} failed: {err}"))?;
let news = self.drain_news(ctx, &key).await?;
let role = dominant_role(&news); let role = dominant_role(&news);
// The permission check is load-bearing: this handler's one tool // The permission check is load-bearing: this handler's one tool
@@ -320,16 +395,24 @@ enum InboxNews {
Notified { Notified {
role: Role, role: Role,
notification: Notification, notification: Notification,
/// The account watch the notification arrived on, when the payload
/// names it: what the reply posts as.
account_id: Option<String>,
}, },
} }
/// The triaged payload [`crate::forge::ForgeSource`] ingests: `repo` and /// The triaged payload [`crate::forge::ForgeSource`] ingests: `repo` and
/// `number` ride at the top level for the key function, the rest is the /// `number` ride at the top level for the key function, the rest is the
/// notification and the role its triage earned. /// notification, the role its triage earned, and the account whose feed
/// delivered it.
#[derive(serde::Deserialize)] #[derive(serde::Deserialize)]
struct TriagedPayload { struct TriagedPayload {
role: Role, role: Role,
notification: Notification, notification: Notification,
/// The identity store id of the account watch that polled this. Absent
/// on payloads ingested before it was recorded.
#[serde(default)]
account_id: Option<String>,
} }
/// Decodes one inbox item into its news. A triaged notification is tried /// Decodes one inbox item into its news. A triaged notification is tried
@@ -340,11 +423,23 @@ fn decode_news(item: &InboxItem) -> Result<InboxNews, ForgeError> {
return Ok(InboxNews::Notified { return Ok(InboxNews::Notified {
role: triaged.role, role: triaged.role,
notification: triaged.notification, notification: triaged.notification,
account_id: triaged.account_id,
}); });
} }
decode(item).map(InboxNews::Event) decode(item).map(InboxNews::Event)
} }
/// The account a batch's notifications were delivered as, when they name
/// one: the first is enough, since a thread fed by two accounts' feeds is
/// two identities on one conversation and the first is as good as the
/// second. Plain events name no account.
fn delivery_account(news: &[InboxNews]) -> Option<&str> {
news.iter().find_map(|news| match news {
InboxNews::Notified { account_id, .. } => account_id.as_deref(),
InboxNews::Event(_) => None,
})
}
/// The batch's dominant role, for the prompt's persona framing: /// The batch's dominant role, for the prompt's persona framing:
/// implementing subsumes reviewing — a batch that asks for both is a /// implementing subsumes reviewing — a batch that asks for both is a
/// dispatch where the work is the point. Plain events imply no persona: /// dispatch where the work is the point. Plain events imply no persona:
@@ -713,10 +808,12 @@ mod tests {
let review = InboxNews::Notified { let review = InboxNews::Notified {
role: Role::Reviewer, role: Role::Reviewer,
notification: notification(NotificationReason::ReviewRequested), notification: notification(NotificationReason::ReviewRequested),
account_id: None,
}; };
let work = InboxNews::Notified { let work = InboxNews::Notified {
role: Role::Implementer, role: Role::Implementer,
notification: notification(NotificationReason::Assigned), notification: notification(NotificationReason::Assigned),
account_id: None,
}; };
assert_eq!( assert_eq!(
dominant_role(std::slice::from_ref(&review)), dominant_role(std::slice::from_ref(&review)),
@@ -741,6 +838,7 @@ mod tests {
&[InboxNews::Notified { &[InboxNews::Notified {
role: Role::Reviewer, role: Role::Reviewer,
notification: notification(NotificationReason::ReviewRequested), notification: notification(NotificationReason::ReviewRequested),
account_id: None,
}], }],
Path::new("/tmp/w"), Path::new("/tmp/w"),
Some(Role::Reviewer), Some(Role::Reviewer),
@@ -825,6 +923,81 @@ mod tests {
); );
} }
/// A thread an account's notification feed delivered has no repo watch
/// row of its own — the account watch covers the whole instance — and
/// the dispatch still resolves a client, as that account.
#[test]
fn an_account_watch_resolves_the_client_a_repo_watch_would_have() {
let watches = Arc::new(ClawStore::in_memory().expect("store opens"));
watches
.add_account_watch("gitea", Some("https://git.example"), "acct-1")
.expect("account watch lands");
let accounts = Arc::new(rhema_identity::VolatileIdentityStore::new());
accounts
.put(
rhema_identity::AccountDraft::new(
"kosmos",
"gitea",
rhema_identity::Secret::new(rhema_identity::SecretKind::Token, "tok"),
)
.with_url("https://git.example")
.with_username("kosmos"),
)
.expect("account lands");
let registry = Arc::new(
crate::backends::BackendRegistry::new(
vec![crate::backends::Backend::headless(
"mock",
Arc::new(NoopProvider),
)],
None,
)
.expect("constructs"),
);
let handler = ClawHandler::new(
registry,
"/tmp/w",
watches,
accounts,
true,
std::collections::BTreeMap::new(),
);
let keys = |handler: &ClawHandler| -> Vec<String> {
handler
.clients
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.keys()
.cloned()
.collect()
};
let key = ResourceKey::from("gitea:issue:acme/widgets#7");
handler
.watch_forge(&key)
.expect("the account watch answers for the thread");
let cached = keys(&handler);
assert_eq!(cached.len(), 1);
assert!(
cached[0].contains("kosmos"),
"the account's login rides the client: {cached:?}"
);
handler.watch_forge(&key).expect("it resolves again");
assert_eq!(keys(&handler).len(), 1, "the second dispatch is cached");
assert!(
handler.watch_forge_as(&key, Some("acct-2")).is_err(),
"naming another account refuses: there is no watch to post as it"
);
assert!(
handler
.watch_forge(&ResourceKey::from("github:issue:acme/widgets#7"))
.is_err(),
"a forge with no watch at all still refuses"
);
}
#[test] #[test]
fn the_transcript_setting_reaches_the_dispatch_config() { fn the_transcript_setting_reaches_the_dispatch_config() {
let watches = Arc::new(ClawStore::in_memory().expect("store opens")); let watches = Arc::new(ClawStore::in_memory().expect("store opens"));