feat(bots): cron jobs post into Group Chats; rooms get a purpose and shared notes

A scheduled job can now deliver to a Bot Mode Group Chat with
`deliver: room:<room_id>[@<profile>]`. The output lands as one message in
the room, labelled with the job's name. It starts a round only for the
members it @mentions (`@all` addresses everyone); without a mention it is
just a message in the room, so a feed of scheduled results never makes
every Bot answer it. Failure notices never start a round. The post is an
ordinary `message.user` event with a server-minted `cron:<job>` actor, so
the existing thread/round machinery and replay stay unchanged and a client
cannot impersonate a job.

Rooms also gain a short purpose and a small shared notes block
(`groups.context`, shown and edited in the room header's Room settings).
Both are rendered at the top of each member's turn prompt, which is the
user-role message, never the system prompt, so prompt caching is
untouched. They are stored as `room.context_set` events in the room log
and read as of the message that opened a discussion: the planner stays a
pure function of the log, and an edit mid-discussion never re-plans a turn
already admitted (it applies from the next message). Rooms without a
purpose keep byte-identical prompts.
This commit is contained in:
Teknium committed 2026-11-03 10:28:21 -08:00
1 parent ac76ec4f88
commit 7fa5bbc71c
22 files changed
+710 -36

No files matched your search

@@ -3,18 +3,21 @@ import { useState } from 'react'
import { CanonicalGroupRoster } from './canonical-group-activity'
import { useCanonicalGroupLabels } from './canonical-group-labels'
import type { CanonicalRoom } from './canonical-groups'
import type { CanonicalRoom, CanonicalRoomContext } from './canonical-groups'
import type { MemberStatus } from './group-room-status'
import { RoomHeader } from './group-room-ui'
interface CanonicalGroupHeaderProps {
busy: boolean
canStop: boolean
context?: CanonicalRoomContext
deliveryTarget: string
detailsOpen: boolean
onBack?: () => void
onDetails: () => void
onDisband: () => Promise<void>
onRename: (name: string) => void
onSaveContext: (context: CanonicalRoomContext) => void
onStop: () => void
room?: CanonicalRoom
statuses: Map<string, MemberStatus>
@@ -22,9 +25,10 @@ interface CanonicalGroupHeaderProps {
}
/** Title (opens the details panel), each member's live chip, and quiet icon actions. */
export function CanonicalGroupHeader({ busy, canStop, detailsOpen, onBack, onDetails, onDisband, onRename, onStop, room, statuses, working }: CanonicalGroupHeaderProps) {
export function CanonicalGroupHeader({ busy, canStop, context, deliveryTarget, detailsOpen, onBack, onDetails, onDisband, onRename, onSaveContext, onStop, room, statuses, working }: CanonicalGroupHeaderProps) {
const labels = useCanonicalGroupLabels()
const [renaming, setRenaming] = useState<null | string>(null)
const [settings, setSettings] = useState<CanonicalRoomContext | null>(null)
const [ending, setEnding] = useState(false)
const rename = (name: string) => {
@@ -44,16 +48,32 @@ export function CanonicalGroupHeader({ busy, canStop, detailsOpen, onBack, onDet
<RoomHeader backLabel={labels.back} onBack={onBack} onStop={busy || !canStop ? undefined : onStop} onTitle={onDetails}
stopLabel={labels.stopGroup} title={room?.name || labels.loadingGroup} titleExpanded={detailsOpen} titleLabel={labels.groupDetails} trailing={<>
{action(labels.rename, 'edit', () => setRenaming(room?.name ?? ''), busy || !room)}
{action(labels.roomSettings, 'settings-gear', () => setSettings({ purpose: context?.purpose ?? '', notes: context?.notes ?? '' }), busy || !room || !context)}
{action(labels.disband, 'trash', () => setEnding(true), busy || !room)}
</>}
working={working}>
<CanonicalGroupRoster label={labels.members} labels={labels} members={room?.members} statuses={statuses} />
</RoomHeader>
{context?.purpose && <p className="truncate px-3 text-xs text-(--ui-text-secondary)" title={context.purpose}>{context.purpose}</p>}
{renaming !== null && <form className="flex items-center gap-2 px-3 py-2" onSubmit={event => { event.preventDefault(); rename(renaming.trim()) }}>
<Input aria-label={labels.renameLabel} autoFocus onChange={event => setRenaming(event.target.value)} value={renaming} />
<Button disabled={busy || !renaming.trim()} size="sm" type="submit">{labels.save}</Button>
<Button onClick={() => setRenaming(null)} size="sm" type="button" variant="text">{labels.cancel}</Button>
</form>}
{settings !== null && <form aria-label={labels.roomSettings} className="grid gap-2 px-3 py-2" onSubmit={event => { event.preventDefault(); setSettings(null); onSaveContext(settings) }}>
<label className="grid gap-1 text-xs">{labels.purposeLabel}
<Input maxLength={500} onChange={event => setSettings({ ...settings, purpose: event.target.value })} placeholder={labels.purposePlaceholder} value={settings.purpose} />
</label>
<label className="grid gap-1 text-xs">{labels.notesLabel}
<textarea className="min-h-24 rounded border border-(--ui-border) bg-transparent p-2 text-xs" onChange={event => setSettings({ ...settings, notes: event.target.value })}
placeholder={labels.notesPlaceholder} value={settings.notes} />
</label>
<p className="text-xs text-(--ui-text-secondary)">{labels.deliveryTarget} <code data-selectable-text="true">{deliveryTarget}</code></p>
<div className="flex gap-2">
<Button disabled={busy} type="submit">{labels.save}</Button>
<Button onClick={() => setSettings(null)} type="button">{labels.cancel}</Button>
</div>
</form>}
{ending && <ConfirmDialog busyLabel={labels.disbanding} confirmLabel={labels.disband}
description={<span>{labels.disbandDescPrefix}<span className="font-medium text-foreground">{room?.name}</span>{labels.disbandDescSuffix(room?.members?.length ?? 0)}</span>}
destructive doneLabel={labels.disbandDone} onClose={() => setEnding(false)} onConfirm={onDisband} open title={labels.disbandTitle} />}
@@ -50,6 +50,9 @@ function eventAuthor(event: CanonicalGroupEvent, roster: Map<string, CanonicalRo
(isMessage ? event.actor?.id : memberId)
}
// The gateway stamps scheduler posts with a `cron:` / `cron-notice:` actor; a client cannot mint one.
const isScheduledPost = (event: CanonicalGroupEvent) => event.kind === 'message.user' && /^cron(-notice)?:/.test(event.actor?.id ?? '')
function eventTimeMs(timestamp?: number) {
if (typeof timestamp !== 'number') {return undefined}
const at = timestamp * 1000
@@ -226,13 +229,18 @@ const MESSAGE_KINDS = new Set(['message', 'message.user', 'message.member'])
function MessageRow({ author, avatarFor, binding, disabled, event, member, memberId, onReply }: Omit<HistoryRowProps, 'notices' | 'roster'> & {
author?: string; member?: CanonicalRoomMember; memberId?: string
}) {
const labels = useCanonicalGroupLabels()
const isUser = event.actor?.kind !== 'member'
const text = event.payload.text || event.payload.content || ''
const canReply = !!onReply && !!text && (!!member || typeof event.payload.thread_id === 'string')
const face = isUser ? undefined : avatarFor?.(member?.profile || member?.handle || memberId || author || 'bot', ROOM_AVATAR_SIZE)
const badge = isScheduledPost(event)
? <span className="shrink-0 rounded bg-(--ui-surface-2) px-1 text-[0.625rem] text-(--ui-text-secondary)">{labels.scheduledPost}</span>
: undefined
return <RoomMessage actions={canReply ? <ReplyAction author={author ?? ''} event={event} member={isUser ? undefined : member} onReply={onReply!} text={text} /> : undefined}
author={author} avatar={face} isUser={isUser} timestampMs={eventTimeMs(event.created_at)}>
author={author} avatar={face} badge={badge} isUser={isUser} timestampMs={eventTimeMs(event.created_at)}>
<EventBody event={event} />
<EventFiles binding={binding} disabled={disabled} event={event} />
</RoomMessage>
@@ -62,6 +62,13 @@ export interface CanonicalGroupMessages {
rowFailed: string
rowNeedsYou: string
lead: string
roomSettings: string
purposeLabel: string
purposePlaceholder: string
notesLabel: string
notesPlaceholder: string
deliveryTarget: string
scheduledPost: string
}
export const CANONICAL_GROUP_LOCALES = {
@@ -128,7 +135,14 @@ export const CANONICAL_GROUP_LOCALES = {
rowWorking: 'Working',
rowFailed: 'A member failed',
rowNeedsYou: 'Needs you',
lead: 'Lead'
lead: 'Lead',
roomSettings: 'Room settings',
purposeLabel: 'Purpose',
purposePlaceholder: 'What this room is for, in one line',
notesLabel: 'Shared notes',
notesPlaceholder: 'Facts every member should keep in mind',
deliveryTarget: 'Scheduled jobs can post here with deliver:',
scheduledPost: 'Scheduled'
},
ja: {
roomStopped: 'グループが停止しました',
@@ -193,7 +207,14 @@ export const CANONICAL_GROUP_LOCALES = {
rowWorking: '作業中',
rowFailed: 'メンバーが失敗しました',
rowNeedsYou: '対応が必要',
lead: 'リード'
lead: 'リード',
roomSettings: 'ルーム設定',
purposeLabel: '目的',
purposePlaceholder: 'このルームの目的を一行で',
notesLabel: '共有メモ',
notesPlaceholder: '全メンバーが覚えておくべき事項',
deliveryTarget: 'スケジュール済みジョブの投稿先 deliver:',
scheduledPost: 'スケジュール'
},
zh: {
roomStopped: '群组已停止',
@@ -258,7 +279,14 @@ export const CANONICAL_GROUP_LOCALES = {
rowWorking: '工作中',
rowFailed: '有成员失败',
rowNeedsYou: '需要你处理',
lead: '负责人'
lead: '负责人',
roomSettings: '群组设置',
purposeLabel: '用途',
purposePlaceholder: '用一句话说明这个群组的用途',
notesLabel: '共享笔记',
notesPlaceholder: '每位成员都应记住的事项',
deliveryTarget: '定时任务可通过 deliver: 发到这里',
scheduledPost: '定时'
},
'zh-hant': {
roomStopped: '群組已停止',
@@ -323,7 +351,14 @@ export const CANONICAL_GROUP_LOCALES = {
rowWorking: '工作中',
rowFailed: '有成員失敗',
rowNeedsYou: '需要你處理',
lead: '負責人'
lead: '負責人',
roomSettings: '群組設定',
purposeLabel: '用途',
purposePlaceholder: '用一句話說明這個群組的用途',
notesLabel: '共用筆記',
notesPlaceholder: '每位成員都應記住的事項',
deliveryTarget: '排程工作可透過 deliver: 張貼到這裡',
scheduledPost: '排程'
},
ar: {
roomStopped: 'أُوقفت المجموعة',
@@ -388,7 +423,14 @@ export const CANONICAL_GROUP_LOCALES = {
rowWorking: 'يعمل',
rowFailed: 'فشل أحد الأعضاء',
rowNeedsYou: 'بحاجة إليك',
lead: 'القائد'
lead: 'القائد',
roomSettings: 'إعدادات الغرفة',
purposeLabel: 'الغرض',
purposePlaceholder: 'الغرض من هذه الغرفة في سطر واحد',
notesLabel: 'ملاحظات مشتركة',
notesPlaceholder: 'حقائق يجب أن يتذكرها كل عضو',
deliveryTarget: 'يمكن للمهام المجدولة النشر هنا باستخدام deliver:',
scheduledPost: 'مجدول'
},
ru: {
roomStopped: 'Группа остановлена',
@@ -456,7 +498,14 @@ export const CANONICAL_GROUP_LOCALES = {
rowWorking: 'Работает',
rowFailed: 'Ошибка участника',
rowNeedsYou: 'Нужны вы',
lead: 'Ведущий'
lead: 'Ведущий',
roomSettings: 'Настройки комнаты',
purposeLabel: 'Назначение',
purposePlaceholder: 'Для чего эта комната, одной строкой',
notesLabel: 'Общие заметки',
notesPlaceholder: 'Что должен помнить каждый участник',
deliveryTarget: 'Задания по расписанию могут писать сюда через deliver:',
scheduledPost: 'По расписанию'
},
fr: {
roomStopped: 'Groupe arrêté',
@@ -523,7 +572,14 @@ export const CANONICAL_GROUP_LOCALES = {
rowWorking: 'Au travail',
rowFailed: 'Un membre a échoué',
rowNeedsYou: 'Vous êtes attendu',
lead: 'Responsable'
lead: 'Responsable',
roomSettings: 'Paramètres du salon',
purposeLabel: 'Objectif',
purposePlaceholder: 'À quoi sert ce salon, en une ligne',
notesLabel: 'Notes partagées',
notesPlaceholder: 'Ce que chaque membre doit garder en tête',
deliveryTarget: 'Les tâches planifiées peuvent publier ici avec deliver:',
scheduledPost: 'Planifié'
},
de: {
roomStopped: 'Gruppe angehalten',
@@ -591,7 +647,14 @@ export const CANONICAL_GROUP_LOCALES = {
rowWorking: 'Arbeitet',
rowFailed: 'Ein Mitglied ist fehlgeschlagen',
rowNeedsYou: 'Braucht dich',
lead: 'Leitung'
lead: 'Leitung',
roomSettings: 'Raumeinstellungen',
purposeLabel: 'Zweck',
purposePlaceholder: 'Wofür dieser Raum da ist, in einer Zeile',
notesLabel: 'Gemeinsame Notizen',
notesPlaceholder: 'Was jedes Mitglied im Blick behalten soll',
deliveryTarget: 'Geplante Aufgaben können hier mit deliver: posten',
scheduledPost: 'Geplant'
},
es: {
roomStopped: 'Grupo detenido',
@@ -658,6 +721,13 @@ export const CANONICAL_GROUP_LOCALES = {
rowWorking: 'Trabajando',
rowFailed: 'Un miembro falló',
rowNeedsYou: 'Te necesita',
lead: 'Responsable'
lead: 'Responsable',
roomSettings: 'Ajustes de la sala',
purposeLabel: 'Propósito',
purposePlaceholder: 'Para qué sirve esta sala, en una línea',
notesLabel: 'Notas compartidas',
notesPlaceholder: 'Lo que cada miembro debe tener presente',
deliveryTarget: 'Las tareas programadas pueden publicar aquí con deliver:',
scheduledPost: 'Programado'
}
} satisfies Record<string, CanonicalGroupMessages>
@@ -11,7 +11,7 @@ import { forgetCanonicalGroup } from './canonical-group-registry'
import { prepareCanonicalGroupSend, readCanonicalGroupSend, retireCanonicalGroupSend } from './canonical-group-send'
import type { PreparedCanonicalGroupSend } from './canonical-group-send'
import { actCanonicalGroup, canonicalGroupRequest } from './canonical-groups'
import type { CanonicalGroupBinding, CanonicalPendingAction, CanonicalRoom, CanonicalRoomMember } from './canonical-groups'
import type { CanonicalGroupBinding, CanonicalPendingAction, CanonicalRoom, CanonicalRoomContext, CanonicalRoomMember } from './canonical-groups'
import { GroupMentionInput } from './group-chat-parts'
import { RoomBotAvatar } from './group-room-avatar'
import { fill, ROOM_COMPOSER_INPUT_CLASS, RoomComposer, RoomReplyChip } from './group-room-ui'
@@ -19,7 +19,7 @@ import { fill, ROOM_COMPOSER_INPUT_CLASS, RoomComposer, RoomReplyChip } from './
type RoomEvent = CanonicalGroupEvent
interface Attachment { attachment_id?: string; event_id?: string; kind: string; name: string; mime: string; size?: number }
interface RoomReply { member?: CanonicalRoomMember; name: string; quote: string; thread?: string }
interface RoomState { room: CanonicalRoom; driver_status?: CanonicalDriverActivity & { pending_actions?: CanonicalPendingAction[] } }
interface RoomState { room: CanonicalRoom; context?: CanonicalRoomContext | null; driver_status?: CanonicalDriverActivity & { pending_actions?: CanonicalPendingAction[] } }
const sameGroupPayload = (payload: Record<string, unknown>, text: string, attachments: Attachment[], thread?: string) =>
payload.text === text && JSON.stringify(payload.attachments ?? []) === JSON.stringify(attachments) &&
@@ -184,6 +184,9 @@ function CanonicalRoomView({ binding: initialBinding, visible, onBack }: {
const rename = (name: string) =>
void mutate(() => canonicalGroupRequest(binding, 'groups.rename', { room_id: binding.roomId, event_id: randomUuid(), name }))
const saveContext = (context: CanonicalRoomContext) =>
void mutate(() => canonicalGroupRequest(binding, 'groups.context', { room_id: binding.roomId, event_id: randomUuid(), ...context }))
// Rejects so ConfirmDialog stays open with the error; a disbanded room no longer reads back, so leave it.
const end = async () => {
await canonicalGroupRequest(binding, 'groups.disband', { room_id: binding.roomId, cancel_id: randomUuid() })
@@ -236,8 +239,9 @@ function CanonicalRoomView({ binding: initialBinding, visible, onBack }: {
return <section className="flex h-full min-h-0">
<div className="flex min-h-0 min-w-0 flex-1 flex-col">
<CanonicalGroupHeader busy={busy} canStop={!!driver} detailsOpen={detailsOpen} onBack={onBack} onDetails={() => setDetailsOpen(open => !open)}
onDisband={end} onRename={rename} onStop={() => void mutate(() => canonicalGroupRequest(binding, 'groups.stop', { room_id: binding.roomId, cancel_id: randomUuid() }))}
<CanonicalGroupHeader busy={busy} canStop={!!driver} context={state?.context ?? undefined}
deliveryTarget={`room:${binding.roomId}${binding.profile ? `@${binding.profile}` : ''}`} detailsOpen={detailsOpen} onBack={onBack} onDetails={() => setDetailsOpen(open => !open)}
onDisband={end} onRename={rename} onSaveContext={saveContext} onStop={() => void mutate(() => canonicalGroupRequest(binding, 'groups.stop', { room_id: binding.roomId, cancel_id: randomUuid() }))}
room={state?.room} statuses={statuses} working={working} />
{readError && <div className="flex items-center gap-2 px-3 py-1.5 text-xs text-destructive" role="alert">{readError}
<Button onClick={() => void refresh().catch(e => setReadError(String(e)))} size="xs" variant="ghost">{labels.refresh}</Button></div>}
@@ -20,6 +20,11 @@ export interface CanonicalRoomMember {
target?: Record<string, unknown>
}
export interface CanonicalRoomContext {
purpose: string
notes: string
}
export interface CanonicalRoom {
room_id: string
name: string
@@ -117,14 +117,15 @@ const AVATAR_COLUMN = 'w-7 shrink-0'
/** One transcript row: avatar, name and time over the body, like a chat thread. Members'
* bodies read like the main thread's assistant answer (bare markdown); the user's own body
* sits in the main thread's user bubble (`--dt-user-bubble` fill, transcript hairline). */
export function RoomMessage({ actions, author, avatar, children, isUser = false, timestampMs }: {
actions?: ReactNode; author?: ReactNode; avatar?: ReactNode; children: ReactNode; isUser?: boolean; timestampMs?: number
export function RoomMessage({ actions, author, avatar, badge, children, isUser = false, timestampMs }: {
actions?: ReactNode; author?: ReactNode; avatar?: ReactNode; badge?: ReactNode; children: ReactNode; isUser?: boolean; timestampMs?: number
}) {
return <div className="group/room-message flex items-start gap-3 py-2" data-room-message={isUser ? 'user' : 'member'}>
<div className={cn(AVATAR_COLUMN, 'mt-0.5')}>{avatar}</div>
<div className="min-w-0 flex-1">
<div className="flex min-h-5 items-center gap-2">
{author && <span className="truncate text-[0.8125rem] font-semibold text-foreground">{author}</span>}
{badge}
{typeof timestampMs === 'number' && Number.isFinite(timestampMs) &&
<time className="shrink-0 text-[0.6875rem] tabular-nums text-(--ui-text-tertiary)" dateTime={new Date(timestampMs).toISOString()}>{relativeTime(timestampMs)}</time>}
{actions && <div className="ml-auto flex shrink-0 items-center gap-0.5 opacity-0 transition-opacity group-hover/room-message:opacity-100 focus-within:opacity-100">{actions}</div>}
@@ -1632,6 +1632,7 @@ export interface GroupsStateParams {
export interface GroupsStateResult {
room: Room
driver_status?: RoomDriverStatus | null
context?: RoomContext | null
}
/** ``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt (``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``) or the driver's approval action (``kind: approval`` + run/session/approval context), or a stalled room preparation's Retry (``task_id`` ``prepare:<room_id>``, ``member_id`` the members not woken). ``member_activity`` lists live tasks in admission order (absent from older gateways). ``last_error`` is this room's last preparation failure, null once one succeeds. */
export interface RoomDriverStatus {
@@ -1660,6 +1661,11 @@ export interface PeerRouteStatus {
member_id: string
status: string
}
/** ``gateway/hosted_room_context.py::room_context``: the purpose and shared notes every member's turn prompt carries (inside the user-role message). */
export interface RoomContext {
purpose: string
notes: string
}
export interface GroupsSendParams {
profile?: string | null
room_id: string
@@ -1681,6 +1687,17 @@ export interface GroupsRenameParams {
export interface GroupsRenameResult {
room: Room
}
export interface GroupsContextParams {
profile?: string | null
room_id: string
event_id: string
purpose: string
notes: string
}
export interface GroupsContextResult {
context: RoomContext
event: RoomEvent
}
export interface GroupsLogParams {
profile?: string | null
room_id: string
@@ -6184,6 +6201,8 @@ export interface RpcMethods {
'groups.attachment.upload': { params: GroupsAttachmentUploadParams; result: RoomAttachment }
/** Describe the hosted-room protocol implemented by this gateway. */
'groups.capabilities': { params: GroupsCapabilitiesParams; result: GroupsCapabilitiesResult }
/** Set one hosted room's purpose and shared notes; members see them from the next message. */
'groups.context': { params: GroupsContextParams; result: GroupsContextResult }
/** Create a hosted room idempotently; authority is this gateway's stable install identity. */
'groups.create': { params: GroupsCreateParams; result: GroupsCreateResult }
/** Fence this gateway's stale room authority against a proven newer epoch. */
@@ -6640,6 +6659,7 @@ export const RPC_METHODS = [
'groups.attachment.download',
'groups.attachment.upload',
'groups.capabilities',
'groups.context',
'groups.create',
'groups.demote',
'groups.disband',
@@ -1140,6 +1140,24 @@
}
}
},
{
"name": "groups.context",
"summary": "Set one hosted room's purpose and shared notes; members see them from the next message.",
"params": [
{
"name": "params",
"schema": {
"$ref": "#/components/schemas/GroupsContextParams"
}
}
],
"result": {
"name": "result",
"schema": {
"$ref": "#/components/schemas/GroupsContextResult"
}
}
},
{
"name": "groups.create",
"summary": "Create a hosted room idempotently; authority is this gateway's stable install identity.",
@@ -19125,6 +19143,64 @@
"title": "GroupsCapabilitiesResult",
"type": "object"
},
"GroupsContextParams": {
"additionalProperties": false,
"properties": {
"profile": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"title": "Profile"
},
"room_id": {
"title": "Room Id",
"type": "string"
},
"event_id": {
"title": "Event Id",
"type": "string"
},
"purpose": {
"title": "Purpose",
"type": "string"
},
"notes": {
"title": "Notes",
"type": "string"
}
},
"required": [
"room_id",
"event_id",
"purpose",
"notes"
],
"title": "GroupsContextParams",
"type": "object"
},
"GroupsContextResult": {
"additionalProperties": false,
"properties": {
"context": {
"$ref": "#/components/schemas/RoomContext"
},
"event": {
"$ref": "#/components/schemas/RoomEvent"
}
},
"required": [
"context",
"event"
],
"title": "GroupsContextResult",
"type": "object"
},
"GroupsCreateParams": {
"additionalProperties": false,
"properties": {
@@ -20334,6 +20410,17 @@
}
],
"default": null
},
"context": {
"anyOf": [
{
"$ref": "#/components/schemas/RoomContext"
},
{
"type": "null"
}
],
"default": null
}
},
"required": [
@@ -34144,6 +34231,26 @@
"title": "RoomAuthority",
"type": "object"
},
"RoomContext": {
"additionalProperties": false,
"description": "``gateway/hosted_room_context.py::room_context``: the purpose and shared notes every member's turn\nprompt carries (inside the user-role message).",
"properties": {
"purpose": {
"title": "Purpose",
"type": "string"
},
"notes": {
"title": "Notes",
"type": "string"
}
},
"required": [
"purpose",
"notes"
],
"title": "RoomContext",
"type": "object"
},
"RoomDriverStatus": {
"additionalProperties": false,
"description": "``HostedRoomService.status(room_id)``; ``pending_actions`` rows are the exact attempt\n(``hosted_room_service.attempt_action``: ``kind`` + ``member_id``/``task_id``/``execution_generation``)\nor the driver's approval action (``kind: approval`` + run/session/approval context), or a stalled\nroom preparation's Retry (``task_id`` ``prepare:<room_id>``, ``member_id`` the members not woken).\n``member_activity`` lists live tasks in admission order (absent from older gateways).\n``last_error`` is this room's last preparation failure, null once one succeeds.",
+12 -2
View File
@@ -1,4 +1,4 @@
"""Cron delivery: target resolution (origin/home/explicit/bot-chat), transcript mirroring and
"""Cron delivery: target resolution (origin/home/explicit/bot-chat/room), transcript mirroring and
session seeding, live-adapter / relay / standalone send lanes, and ``_deliver_result``.
Split out of ``cron.scheduler``. Import names from this module directly (``cron.scheduler`` only
@@ -602,6 +602,9 @@ def _resolve_single_delivery_target(
origin = _resolve_origin(job)
if deliver_value == "local":
return None
room_address = parse_room_deliver_token(deliver_value)
if room_address is not None:
return resolve_room_target(job, room_address)
# Must precede the generic platform:chat_id split so the profile name isn't parsed as chat_id.
bot_chat_profile = parse_bot_chat_deliver_token(deliver_value)
if bot_chat_profile is not None:
@@ -1796,7 +1799,7 @@ def _deliver_result(
external_execution = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER", "")
if (external_execution and adapters is None
and external_execution == str(job.get("execution_id") or "")
and any(target["platform"] != BOT_CHAT_PLATFORM for target in targets)):
and any(target["platform"] not in (BOT_CHAT_PLATFORM, ROOM_PLATFORM) for target in targets)):
from cron.delivery_queue import enqueue_and_wait
_record_delivery_verification(job, [])
@@ -1900,6 +1903,11 @@ def _deliver_result(
if receipt and receipt["status"] == "ambiguous":
unverified_targets.append(bot_chat_error)
continue
if target["platform"] == ROOM_PLATFORM:
room_error = deliver_to_room(job, content, target["chat_id"], for_failure=for_failure)
if room_error:
delivery_errors.append(room_error)
continue
if target["platform"] == "local":
from gateway.session_cron import deliver_to_local_chat
from hermes_constants import get_hermes_home
@@ -1938,6 +1946,8 @@ def _deliver_result(
# Late-bound origin namespace (see module docstring). Imported LAST so this module is fully
# populated before ``scheduler`` re-exports from it.
from cron import scheduler as _sched
from cron.scheduler_delivery_room import (
ROOM_PLATFORM, deliver_to_room, parse_room_deliver_token, resolve_room_target)
from cron import scheduler_delivery_origin as _origin
from cron import scheduler_preflight as _preflight
from cron import scheduler_script as _script
+85
View File
@@ -0,0 +1,85 @@
"""Cron delivery into a Bot Mode Group Chat (``deliver: room:<room_id>[@<profile>]``).
The job's output becomes one message in the room's log, labelled with the job's name. It starts a round
only for the members it @mentions (``@all`` addresses everyone); without a mention it is only a message in
the room, so a feed of scheduled results never makes every Bot answer it. Failure notices never start a
round. The post is appended to the hosting gateway's room store directly; the room worker picks it up on
its next poll, so delivery does not need the Desktop open.
"""
from __future__ import annotations
import logging
import re
import uuid
from pathlib import Path
from typing import Optional
logger = logging.getLogger(__name__)
ROOM_PLATFORM = "room"
_PREFIX = ROOM_PLATFORM + ":"
_ROOM_TOKEN_RE = re.compile(r"^(?P<room>[A-Za-z0-9][A-Za-z0-9._:-]*)(?:@(?P<profile>[A-Za-z0-9_-]+))?$")
def parse_room_deliver_token(part: str) -> Optional[str]:
"""``room:<id>[@<profile>]`` → ``"<id>[@<profile>]"``, ``""`` for a malformed room token, or None
when ``part`` is not a room token at all."""
raw = (part or "").strip()
if not raw.lower().startswith(_PREFIX):
return None
rest = raw[len(_PREFIX):].strip()
return rest if _ROOM_TOKEN_RE.fullmatch(rest) else ""
def resolve_room_target(job: dict, address: str) -> Optional[dict]:
if not address:
logger.warning("Job '%s': malformed room delivery target (expected room:<room_id>[@<profile>])",
job.get("id", "?"))
return None
return {"platform": ROOM_PLATFORM, "chat_id": address, "thread_id": None}
def _room_stores(profile: Optional[str]) -> list[Path]:
"""Where a room can live: the hosting profile's session store (rooms the gateway owns), then the
install's shared room store (rooms the dashboard backend owns)."""
from gateway.hosted_rooms import default_db_path
from hermes_constants import get_hermes_home
if profile:
from hermes_cli.profiles import get_profile_dir
home = get_profile_dir(profile)
else:
home = get_hermes_home()
return [home / "state.db", default_db_path()]
def _find_room_store(room_id: str, profile: Optional[str]) -> Optional[Path]:
from gateway.hosted_rooms import probe_hosted_room
return next((path for path in _room_stores(profile) if probe_hosted_room(path, room_id=room_id)), None)
def deliver_to_room(job: dict, content: str, address: str, *, for_failure: bool = False) -> Optional[str]:
"""Post one cron output into a hosted room. None on success, else an error string."""
from cron.scheduler_delivery import _redact_cron_payload
from gateway.hosted_room_context import post_scheduled_message
target = f"room:{address}"
match = _ROOM_TOKEN_RE.fullmatch(address)
if match is None:
return f"{target}: malformed room delivery target"
room_id, profile = match.group("room"), match.group("profile")
try:
store = _find_room_store(room_id, profile)
if store is None:
return f"{target}: no live Group Chat with that id on this machine"
job_id = str(job.get("id") or "job")
# The durable execution id makes a re-delivered run idempotent; direct helper calls get a fresh key.
run_key = str(job.get("execution_id") or job.setdefault("_room_run_key", uuid.uuid4().hex))
event = post_scheduled_message(
store, room_id=room_id, job_id=job_id,
job_name=_redact_cron_payload(str(job.get("name") or job_id), "job name"),
run_key=run_key, text=_redact_cron_payload(content, "room payload"), notice=for_failure)
logger.info("Job '%s': posted to Group Chat %s (seq %s)", job_id, room_id, event.get("seq"))
return None
except Exception as exc:
return f"{target} delivery failed: {exc}"
+2
View File
@@ -268,6 +268,8 @@ def _preflight_check_delivery(job: dict) -> Optional[str]:
# bot-chat targets deliver via a local subprocess; failures land in last_delivery_error.
if _delivery.parse_bot_chat_deliver_token(part) is not None:
continue
if _delivery.parse_room_deliver_token(part) is not None:
continue # Group Chat posts write the room store directly; no adapter needed.
platform_parts.append(part.split(":", 1)[0].strip())
if not platform_parts:
return None
+114
View File
@@ -0,0 +1,114 @@
"""Room purpose, shared notes and scheduled-job posts for hosted Group Chats.
Both live in the append-only room log, never in a side file: the Discussion planner must stay a pure
function of durable room state, because a member turn's task id hashes its prompt. Purpose and notes are
read as of the message that opened a discussion, so editing them mid-discussion never re-plans a turn that
is already admitted; the edit applies from the next message.
Scheduled posts are ordinary ``message.user`` events whose server-owned actor id names the cron job. Only
this module mints that actor (``groups.send`` hard-codes ``desktop``), so a client cannot impersonate a job.
"""
from __future__ import annotations
import hashlib
import json
import re
from collections.abc import Mapping
from contextlib import closing
from typing import Any
from gateway import hosted_rooms
from gateway.hosted_rooms_common import DbPath
MAX_PURPOSE_CHARS = 500
MAX_NOTES_BYTES = 4 * 1024
CONTEXT_EVENT_KIND = "room.context_set"
_CONTEXT_ACTOR = {"kind": "system", "id": "room-control"}
# Actor id prefixes for posts the scheduler makes. A result post can start a round when it @mentions a
# member; a failure notice never does (its text is machine-written, not addressed to anyone).
SCHEDULED_POST_PREFIX = "cron:"
SCHEDULED_NOTICE_PREFIX = "cron-notice:"
_EMPTY_CONTEXT = {"purpose": "", "notes": ""}
def validate_context(purpose: Any, notes: Any) -> dict[str, str]:
"""Normalize one purpose/notes pair; both may be empty (clears them)."""
if not isinstance(purpose, str) or not isinstance(notes, str):
raise hosted_rooms.HostedRoomError("purpose and notes must be strings")
purpose = " ".join(purpose.split())
notes = notes.strip()
if len(purpose) > MAX_PURPOSE_CHARS:
raise hosted_rooms.HostedRoomError(f"purpose must be at most {MAX_PURPOSE_CHARS} characters")
if len(notes.encode("utf-8")) > MAX_NOTES_BYTES:
raise hosted_rooms.HostedRoomError(f"notes must be at most {MAX_NOTES_BYTES} bytes")
return {"purpose": purpose, "notes": notes}
def _local_authority(db_path: DbPath, room_id: Any) -> dict[str, Any]:
state = hosted_rooms.room_state(db_path, room_id=room_id)
if state["authority_gateway_id"] != hosted_rooms.local_authority_gateway_id():
raise hosted_rooms.AuthorityConflictError("This Group Chat is managed by another gateway.")
return state
def room_context(db_path: DbPath, *, room_id: Any, before_seq: int | None = None) -> dict[str, str]:
"""The latest purpose/notes, or the ones in force before ``before_seq`` when given."""
room_id = hosted_rooms._room_id(room_id)
sql = "SELECT payload_json FROM hosted_room_events WHERE room_id=? AND kind=?"
params: tuple[Any, ...] = (room_id, CONTEXT_EVENT_KIND)
if before_seq is not None:
sql, params = sql + " AND seq<?", (*params, int(before_seq))
with closing(hosted_rooms._read_connection(db_path)) as conn:
row = conn.execute(sql + " ORDER BY seq DESC LIMIT 1", params).fetchone()
if row is None:
return dict(_EMPTY_CONTEXT)
payload = json.loads(row["payload_json"])
return {key: str(payload.get(key) or "") for key in _EMPTY_CONTEXT}
def set_room_context(
db_path: DbPath, *, room_id: Any, event_id: Any, purpose: Any, notes: Any, now: float | None = None
) -> dict[str, Any]:
"""Record a new purpose/notes pair as one replayable room event."""
context = validate_context(purpose, notes)
state = _local_authority(db_path, room_id)
event = hosted_rooms.append_event(
db_path, room_id=room_id, event_id=event_id, kind=CONTEXT_EVENT_KIND, actor=dict(_CONTEXT_ACTOR),
payload=context, authority_gateway_id=state["authority_gateway_id"],
authority_epoch=state["authority_epoch"], now=now)
return {"context": context, "event": event}
def scheduled_post_kind(actor: Mapping[str, Any] | None) -> str | None:
"""``"post"``/``"notice"`` for a scheduler-authored ``message.user`` actor, else None (the human)."""
if not isinstance(actor, Mapping) or actor.get("kind") != "user":
return None
actor_id = str(actor.get("id") or "")
if actor_id.startswith(SCHEDULED_NOTICE_PREFIX):
return "notice"
return "post" if actor_id.startswith(SCHEDULED_POST_PREFIX) else None
def post_scheduled_message(
db_path: DbPath, *, room_id: Any, job_id: str, job_name: str, run_key: str, text: str, notice: bool = False,
now: float | None = None,
) -> dict[str, Any]:
"""Append one cron output to a room. Idempotent per ``run_key``: a retried delivery of the same run
returns the original event instead of posting twice."""
from gateway.hosted_room_discussion import MAX_USER_TEXT_BYTES, validate_user_payload
state = _local_authority(db_path, room_id)
digest = hashlib.sha256("\0".join((str(state["room_id"]), job_id, run_key, str(notice))).encode()).hexdigest()
suffix = "\n\n[Output truncated to fit the room.]"
if len(text.encode("utf-8")) > MAX_USER_TEXT_BYTES:
text = text.encode("utf-8")[:MAX_USER_TEXT_BYTES - len(suffix.encode())].decode("utf-8", "ignore") + suffix
payload = validate_user_payload({"text": text, "thread_id": f"cron-{digest[:24]}"})
prefix = SCHEDULED_NOTICE_PREFIX if notice else SCHEDULED_POST_PREFIX
job_label = re.sub(r"[^A-Za-z0-9._:-]", "-", job_id)[: hosted_rooms.MAX_ACTOR_ID_CHARS - len(prefix)] or "job"
actor = {"kind": "user", "id": f"{prefix}{job_label}",
"display_name": (job_name or job_id)[: hosted_rooms.MAX_ACTOR_LABEL_CHARS]}
return hosted_rooms.append_event(
db_path, room_id=room_id, event_id=f"cron-post:{digest[:40]}", kind="message.user", actor=actor,
payload=payload, authority_gateway_id=state["authority_gateway_id"],
authority_epoch=state["authority_epoch"], now=now)
+36 -6
View File
@@ -19,6 +19,7 @@ from agent.prompt_builder import CONTROL_FRAME_OPENERS
from gateway import hosted_room_driver as driver
from gateway import hosted_rooms
from gateway import hosted_rooms_common as common
from gateway.hosted_room_context import scheduled_post_kind, validate_context
from gateway.hosted_room_task_input import validate_task_input
from gateway.hosted_rooms_common import compact_json
@@ -557,7 +558,10 @@ def _rotate(members: Sequence[DiscussionMember], round_index: int) -> tuple[Disc
def _format_message(event: _ValidatedEvent, room: DiscussionRoom) -> str:
if event.kind == "message.user":
return f"User (user): {event.payload['text']}"
if scheduled_post_kind(event.actor) is None:
return f"User (user): {event.payload['text']}"
job = event.actor.get("display_name") or event.actor.get("id")
return f'Scheduled job "{job}" (automated post, not the user): {event.payload["text"]}'
text = _MEMBER_CONTROL_FRAME_RE.sub(_MEMBER_CONTROL_FRAME_RELABEL, event.payload["text"])
return f"@{_member_by_id(room, event.payload['member_id']).handle}: {text}"
@@ -571,14 +575,25 @@ def _truncate_utf8_text(value: Any, *, max_bytes: int, suffix: str = "") -> str:
return prefix + suffix if prefix else suffix.strip()
def _context_lines(context: Mapping[str, str] | None) -> list[str]:
"""Room purpose and shared notes, rendered inside the user-role turn prompt (never the system prompt)."""
lines: list[str] = []
if context and context.get("purpose"):
lines.append(f"Room purpose (set by the user): {context['purpose']}")
if context and context.get("notes"):
lines += ["Room notes (shared with every member; the user keeps them in room settings):",
*(f" {line}" for line in context["notes"].splitlines())]
return [*lines, ""] if lines else []
def _build_prompt(
*, room: DiscussionRoom, member: DiscussionMember, messages: Sequence[_ValidatedEvent], watermark: int,
seen_through_seq: int, reply_required: bool = False) -> str:
seen_through_seq: int, reply_required: bool = False, context: Mapping[str, str] | None = None) -> str:
delta = [event for event in messages if watermark < event.seq <= seen_through_seq][- MAX_DISCUSSION_DELTA_LINES:]
peers = ", ".join(f"@{candidate.handle}" for candidate in room.members if candidate.member_id != member.member_id)
opening = [
f'[Discussion: "{room.name}"] You are @{member.handle}, one participant '
f"with {peers or 'no other members'} and the user.", "",
f"with {peers or 'no other members'} and the user.", "", *_context_lines(context),
"New messages in this thread since your last turn (oldest first):"]
participation = [
"- The user directly addressed you. Reply with your answer, result, or concrete blocker.",
@@ -722,14 +737,29 @@ def _bounded_task_delta(messages: Sequence[_ValidatedEvent], *, watermark: int,
return selected, attachments
def _first_round_responders(discussion: _ValidatedEvent, room: DiscussionRoom) -> tuple[DiscussionMember, ...]:
"""The human's message with no mention addresses everyone. A scheduled post is only a message in the
room: it pulls in exactly the members it @mentions, and a failure notice pulls in nobody."""
scheduled = scheduled_post_kind(discussion.actor)
if scheduled == "notice":
return ()
return resolve_mentions((str(discussion.payload["text"]),), room.members, default_all=scheduled is None)
def plan_next_task(
room_value: Any, events: Sequence[Mapping[str, Any]], *, local_profiles: Iterable[str],
initial_watermarks: Mapping[tuple[str, str], int] | None = None,
freeze_input_context: bool = False,
limits: DiscussionLimits = DiscussionLimits(),
room_context: Mapping[str, Any] | None = None,
) -> DiscussionDecision:
"""Replay the complete room log and return at most one next member task."""
"""Replay the complete room log and return at most one next member task. ``room_context`` is the
purpose/notes in force when the pending discussion opened (``hosted_room_context.room_context``)."""
room = validate_room(room_value, local_profiles=local_profiles)
try:
context = validate_context(**{key: (room_context or {}).get(key, "") for key in ("purpose", "notes")})
except ValueError as exc:
raise DiscussionValidationError(str(exc)) from exc
validated = _validated_events(events, room=room)
if (discussion := _pending_discussion(validated)) is None:
return DiscussionDecision(status="idle", reason="no_pending_user_event")
@@ -751,7 +781,7 @@ def plan_next_task(
# watermark remains intact, so a peer cited later still receives the
# complete bounded transcript delta without consuming turns meanwhile.
responders = (
resolve_mentions((str(discussion.payload["text"]),), room.members) if round_index == 0
_first_round_responders(discussion, room) if round_index == 0
else _unaddressed_member_mentions(discussion_messages, room))
for member_index, member in enumerate(_rotate(responders, round_index)):
watermark = watermarks.get((thread_id, member.member_id), 0)
@@ -761,7 +791,7 @@ def plan_next_task(
seen_through_seq = delta[-1].seq
prompt = _build_prompt(
room=room, member=member, messages=thread_messages, watermark=watermark,
seen_through_seq=seen_through_seq,
seen_through_seq=seen_through_seq, context=context,
reply_required=_requires_direct_reply(room, discussion, member, round_index))
return decide("task", "member_turn", task=_make_task_plan(
room=room, discussion_event=discussion, member=member, member_index=member_index,
+2 -2
View File
@@ -53,8 +53,8 @@ _EVENT_KINDS_BY_ACTOR = {
"member.unavailable", "room.activity", "room.stop_requested", "turn.deferred", "turn.reassigned",
"turn.cancelled", "turn.failed", "turn.settled", "turn.started"}),
"system": frozenset({
"authority.claimed", "authority.lost", "room.created", "room.disbanded", "room.members_changed", "room.renamed"
})}
"authority.claimed", "authority.lost", "room.context_set", "room.created", "room.disbanded",
"room.members_changed", "room.renamed"})}
_OPTIONAL_ACTOR_FIELDS = (
("display_name", MAX_ACTOR_LABEL_CHARS), ("profile", MAX_ACTOR_ID_CHARS), ("connection_id", MAX_ACTOR_ID_CHARS))
_ACTOR_FIELDS = frozenset({"kind", "id", *(field for field, _ in _OPTIONAL_ACTOR_FIELDS)})
+8 -2
View File
@@ -17,6 +17,7 @@ GROUP_METHODS = {
'groups.log': 'session:read',
'groups.create': 'session:control',
'groups.rename': 'session:control',
'groups.context': 'session:control',
'groups.disband': 'session:control',
'groups.send': 'session:submit',
'groups.attachment.upload': 'session:submit',
@@ -33,6 +34,7 @@ _FIELDS = {
'groups.log': {'room_id', 'since_seq', 'limit', 'include_disbanded'},
'groups.create': {'room_id', 'name', 'members'},
'groups.rename': {'room_id', 'event_id', 'name'},
'groups.context': {'room_id', 'event_id', 'purpose', 'notes'},
'groups.disband': {'room_id', 'cancel_id'},
'groups.send': {'room_id', 'event_id', 'payload'},
'groups.attachment.upload': {'room_id', 'upload_id', 'kind', 'name', 'mime', 'data_base64'},
@@ -87,6 +89,7 @@ async def dispatch_group_control(connection, method, params):
def _group(authority, actor, method, params, *, host_scope=False):
from gateway import hosted_rooms as rooms
from gateway.hosted_room_context import room_context, set_room_context
db_path = authority.db.db_path
gateway_id = rooms.local_authority_gateway_id()
service = getattr(authority, 'hosted_room_service', None)
@@ -212,8 +215,10 @@ def _group(authority, actor, method, params, *, host_scope=False):
def state():
room = rooms.room_state(db_path, **params)
result = {'room': room}
if service is not None and room.get('disbanded_at') is None:
result['driver_status'] = service.status(room['room_id'])
if room.get('disbanded_at') is None:
result['context'] = room_context(db_path, room_id=room['room_id'])
if service is not None:
result['driver_status'] = service.status(room['room_id'])
return result
handlers = {
@@ -223,6 +228,7 @@ def _group(authority, actor, method, params, *, host_scope=False):
'groups.state': state,
'groups.log': lambda: rooms.read_events(db_path, **params),
'groups.rename': lambda: {'room': rooms.rename_room(db_path, **params)},
'groups.context': lambda: set_room_context(db_path, **params),
'groups.disband': disband,
}
return handlers[method]()
+49
View File
@@ -0,0 +1,49 @@
"""Cron ``deliver: room:<id>`` posts into a real hosted room store under a temp HERMES_HOME."""
from __future__ import annotations
import pytest
from cron import scheduler_delivery as delivery
from gateway import hosted_rooms
from hermes_constants import get_hermes_home
ROOM_ID = "room-cron"
MEMBERS = [{"member_id": f"m-{p}", "profile": p, "handle": p} for p in ("alpha", "beta")]
@pytest.fixture
def room_store():
db = get_hermes_home() / "state.db"
hosted_rooms.create_room(
db, room_id=ROOM_ID, name="Feed", members=MEMBERS,
authority_gateway_id=hosted_rooms.local_authority_gateway_id())
return db
def _job(**extra) -> dict:
return {"id": "j1", "name": "Digest", "deliver": f"room:{ROOM_ID}", "execution_id": "exec-1", **extra}
def test_room_target_resolves_and_posts_job_output(room_store, monkeypatch):
monkeypatch.setattr(delivery._sched, "load_config", lambda: {"cron": {"wrap_response": True}})
assert delivery._resolve_delivery_targets(_job()) == [
{"platform": "room", "chat_id": ROOM_ID, "thread_id": None}]
assert delivery._deliver_result(_job(), "Three new issues today") is None
assert delivery._deliver_result(_job(), "Three new issues today") is None # same run: one post
posts = [e for e in hosted_rooms.read_events(room_store, room_id=ROOM_ID)["events"] if e["kind"] == "message.user"]
assert len(posts) == 1
assert posts[0]["payload"]["text"] == "Three new issues today"
assert posts[0]["actor"] == {"kind": "user", "id": "cron:j1", "display_name": "Digest"}
def test_unknown_room_reports_a_delivery_error(room_store):
error = delivery._deliver_result(_job(deliver="room:nope"), "x")
assert error and "no live Group Chat" in error
def test_malformed_room_token_resolves_to_nothing():
assert delivery._resolve_delivery_targets({"id": "j", "deliver": "room:"}) == []
+106
View File
@@ -0,0 +1,106 @@
"""Group Chat purpose/notes and scheduled-job posts, through the real room store and Discussion policy."""
from __future__ import annotations
from pathlib import Path
import pytest
from gateway import hosted_room_context as context
from gateway import hosted_room_discussion as discussion
from gateway import hosted_rooms
ROOM_ID = "room-ctx"
PROFILES = ("research", "build")
MEMBERS = [{"member_id": f"member-{p}", "profile": p, "handle": p} for p in PROFILES]
@pytest.fixture
def room_db(tmp_path: Path) -> Path:
db = tmp_path / "state.db"
hosted_rooms.create_room(
db, room_id=ROOM_ID, name="Ops", members=MEMBERS,
authority_gateway_id=hosted_rooms.local_authority_gateway_id(), now=1)
return db
def _plan(db: Path, *, before_seq: int | None = None) -> discussion.DiscussionDecision:
room = hosted_rooms.room_state(db, room_id=ROOM_ID)
room["members"] = MEMBERS
events = hosted_rooms.read_events(db, room_id=ROOM_ID, limit=hosted_rooms.MAX_LOG_LIMIT)["events"]
return discussion.plan_next_task(
room, events, local_profiles=PROFILES,
room_context=None if before_seq is None else context.room_context(db, room_id=ROOM_ID, before_seq=before_seq))
def _post(db: Path, text: str, *, run_key: str = "run-1", notice: bool = False) -> dict:
return context.post_scheduled_message(
db, room_id=ROOM_ID, job_id="job42", job_name="Nightly", run_key=run_key, text=text, notice=notice)
def test_scheduled_post_without_mention_is_only_a_message(room_db: Path):
event = _post(room_db, "All backups OK")
assert event["actor"] == {"kind": "user", "id": "cron:job42", "display_name": "Nightly"}
decision = _plan(room_db)
assert (decision.status, decision.reason, decision.task) == ("settled", "silent_round", None)
def test_scheduled_post_mention_pulls_in_only_that_member(room_db: Path):
_post(room_db, "Disk at 91% @build please look")
decision = _plan(room_db)
assert decision.status == "task"
assert decision.task.member.profile == "build"
assert 'Scheduled job "Nightly" (automated post, not the user): Disk at 91%' in decision.task.payload["prompt"]
assert "User (user):" not in decision.task.payload["prompt"]
def test_failure_notice_never_starts_a_round(room_db: Path):
_post(room_db, "job failed: @build @all", notice=True)
assert _plan(room_db).status == "settled"
def test_redelivering_the_same_run_posts_once(room_db: Path):
first, again = _post(room_db, "report"), _post(room_db, "report")
assert again["idempotent"] and again["seq"] == first["seq"]
def test_purpose_and_notes_ride_the_user_turn_prompt_and_freeze_per_discussion(room_db: Path):
context.set_room_context(
room_db, room_id=ROOM_ID, event_id="ctx-1", purpose="Ship the 2.0 release",
notes="Freeze is Friday\nOwner: @build")
message = hosted_rooms.append_event(
room_db, room_id=ROOM_ID, event_id="u1", kind="message.user", actor={"kind": "user", "id": "desktop"},
payload={"text": "status?", "thread_id": "t1"},
authority_gateway_id=hosted_rooms.local_authority_gateway_id(), authority_epoch=1)
first = _plan(room_db, before_seq=message["seq"])
prompt = first.task.payload["prompt"]
assert "Room purpose (set by the user): Ship the 2.0 release" in prompt
assert " Freeze is Friday\n Owner: @build" in prompt
assert prompt.index("Room purpose") < prompt.index("User (user): status?")
# An edit after the discussion opened applies from the next message: the planned turn is unchanged.
context.set_room_context(room_db, room_id=ROOM_ID, event_id="ctx-2", purpose="Something else", notes="")
assert _plan(room_db, before_seq=message["seq"]).task.identity == first.task.identity
assert context.room_context(room_db, room_id=ROOM_ID) == {"purpose": "Something else", "notes": ""}
def test_room_without_context_keeps_the_existing_prompt(room_db: Path):
message = hosted_rooms.append_event(
room_db, room_id=ROOM_ID, event_id="u1", kind="message.user", actor={"kind": "user", "id": "desktop"},
payload={"text": "hi", "thread_id": "t1"},
authority_gateway_id=hosted_rooms.local_authority_gateway_id(), authority_epoch=1)
assert _plan(room_db, before_seq=message["seq"]).task.identity == _plan(room_db).task.identity
def test_context_is_bounded(room_db: Path):
with pytest.raises(hosted_rooms.HostedRoomError):
context.set_room_context(room_db, room_id=ROOM_ID, event_id="c", purpose="x" * 501, notes="")
with pytest.raises(hosted_rooms.HostedRoomError):
context.set_room_context(
room_db, room_id=ROOM_ID, event_id="c", purpose="", notes="x" * (context.MAX_NOTES_BYTES + 1))
+1 -1
View File
@@ -177,7 +177,7 @@ def _mode_guidance_notes(job: dict[str, Any], user_deliver: Optional[str]) -> li
parts = target.strip().split(":")
if (
len(parts) == 2
and parts[0] not in ("bot-chat", "sms")
and parts[0] not in ("bot-chat", "room", "sms")
and parts[1]
and not parts[1].startswith("#")):
notes.append(
+1 -1
View File
@@ -1159,7 +1159,7 @@ Jobs run in a fresh session with no current-chat context, so prompts must be sel
},
"deliver": {
"type": "string",
"description": "Where the job's output is POSTED as a one-way message (the job itself always runs in a fresh session with no chat context). Omit to address the chat/topic this job was created from. Otherwise: 'local' (save only, no delivery), 'all' (every connected home channel, resolved at fire time), 'bot-chat' or 'bot-chat:<profile>' (inject into a Bot Chat as a real message), or platform:chat_id:thread_id (e.g. 'telegram:-1001234567890:17585'). Comma-combine like 'origin,all'. From a local chat (CLI, TUI or Desktop) the default writes the result into that chat's history as a note, seen when the chat is reopened and by its next turn, not as a live message."
"description": "Where the job's output is POSTED as a one-way message (the job itself always runs in a fresh session with no chat context). Omit to address the chat/topic this job was created from. Otherwise: 'local' (save only, no delivery), 'all' (every connected home channel, resolved at fire time), 'bot-chat' or 'bot-chat:<profile>' (inject into a Bot Chat as a real message), 'room:<room_id>' (post into a Bot Mode Group Chat; it starts a round only for members the output @mentions), or platform:chat_id:thread_id (e.g. 'telegram:-1001234567890:17585'). Comma-combine like 'origin,all'. From a local chat (CLI, TUI or Desktop) the default writes the result into that chat's history as a note, seen when the chat is reopened and by its next turn, not as a live message."
},
"failure_deliver": {
"type": "string",
+24
View File
@@ -248,9 +248,18 @@ class RoomDriverStatus(Result):
peer_routes: list[PeerRouteStatus]
class RoomContext(Result):
"""``gateway/hosted_room_context.py::room_context``: the purpose and shared notes every member's turn
prompt carries (inside the user-role message)."""
purpose: str
notes: str
class GroupsStateResult(Result):
room: Room
driver_status: RoomDriverStatus | None = None
context: RoomContext | None = None
method("groups.state", params=GroupsStateParams, result=GroupsStateResult,
@@ -289,6 +298,21 @@ method("groups.rename", params=GroupsRenameParams, result=GroupsRenameResult,
doc="Rename one hosted room atomically with its replay event.")
class GroupsContextParams(RoomParams):
event_id: str
purpose: str
notes: str
class GroupsContextResult(Result):
context: RoomContext
event: RoomEvent
method("groups.context", params=GroupsContextParams, result=GroupsContextResult,
doc="Set one hosted room's purpose and shared notes; members see them from the next message.")
class GroupsLogParams(RoomParams):
since_seq: int | None = None
limit: int | None = None
+7 -1
View File
@@ -18,6 +18,7 @@ from gateway import hosted_room_discussion as discussion
from gateway import hosted_room_driver as driver
from gateway import hosted_room_links
from gateway import hosted_rooms
from gateway.hosted_room_context import room_context
from gateway.hosted_room_policy_checkpoint import HostedRoomPolicyCheckpoint, PolicySnapshot
from gateway.hosted_room_peer import (
GatewayRoomCatalog, HostedMemberDispatch, PROTOCOL_VERSION, room_grant_needs_dispatch_refresh)
@@ -495,9 +496,14 @@ class HostedRoomService:
self.db_path, room_id=binding.room_id, clock=self.runtime.clock)
if next(iter(self._list_tasks(binding.room_id, _LIVE_STATUSES)), None) is not None:
return
# The snapshot holds one active discussion; its latest user message opened it.
source_seq = max(
(int(event["seq"]) for event in snapshot.events if event.get("kind") == "message.user"), default=None)
decision = discussion.plan_next_task(
room, list(snapshot.events), local_profiles=self.room_profiles(room),
initial_watermarks=snapshot.watermarks, freeze_input_context=True, limits=self.discussion_limits)
initial_watermarks=snapshot.watermarks, freeze_input_context=True, limits=self.discussion_limits,
room_context=None if source_seq is None else room_context(
self.db_path, room_id=binding.room_id, before_seq=source_seq))
if decision.status == "task" and decision.task is not None:
existing = driver.get_task_for_turn(self.db_path, decision.task.identity)
legacy_payload = dict(decision.task.payload)
+10 -3
View File
@@ -20,7 +20,7 @@ method = _registry.method
#: Wire order of ``groups.capabilities.methods``; every one runs on the RPC pool.
_METHODS = (
"groups.capabilities", "groups.list", "groups.create", "groups.state", "groups.send",
"groups.rename", "groups.log", "groups.disband", "groups.replicate", "groups.replica_state",
"groups.rename", "groups.context", "groups.log", "groups.disband", "groups.replicate", "groups.replica_state",
"groups.promote", "groups.demote", "groups.stop", "groups.retry", "groups.approve",
"groups.peer.invite", "groups.peer.revoke", "groups.peer.register", "groups.attachment.upload",
"groups.attachment.download")
@@ -384,8 +384,11 @@ def _(rid, params: dict, db_path) -> dict:
include_disbanded=params.get("include_disbanded") is True)
service = get_hosted_room_service()
result = {"room": room}
if service is not None and room.get("disbanded_at") is None:
result["driver_status"] = service.status(str(room["room_id"]))
if room.get("disbanded_at") is None:
from gateway.hosted_room_context import room_context
result["context"] = room_context(db_path, room_id=room["room_id"])
if service is not None:
result["driver_status"] = service.status(str(room["room_id"]))
return _ok(rid, result)
@@ -524,6 +527,10 @@ _passthrough(
"groups.rename", "gateway.hosted_rooms", "rename_room",
"""Rename one hosted room atomically with its replay event.""",
code=5117, room_code=4117, params=("room_id", "event_id", "name"), wrap="room")
_passthrough(
"groups.context", "gateway.hosted_room_context", "set_room_context",
"""Set one hosted room's purpose and shared notes (both included in every member's turn prompt).""",
code=5121, room_code=4125, params=("room_id", "event_id", "purpose", "notes"))
_passthrough(
"groups.log", "gateway.hosted_rooms", "read_events",
"""Return a monotonic room-log delta after ``since_seq``.""",