diff --git a/app/src/components/intelligence/AttentionQueueItem.test.tsx b/app/src/components/intelligence/AttentionQueueItem.test.tsx index 17f9c9552..c2791cbce 100644 --- a/app/src/components/intelligence/AttentionQueueItem.test.tsx +++ b/app/src/components/intelligence/AttentionQueueItem.test.tsx @@ -16,6 +16,19 @@ const approval: AttentionItem = { createdAt: '2026-07-06T10:00:00Z', }; +// A REMOTE agent session parked on an approval: same `approval` kind as a local +// gate approval, but its action opens the orchestration chat window instead of +// the local approval surface. +const remoteApproval: AttentionItem = { + id: 'remote-approval:h-remote', + kind: 'approval', + instanceId: 'h-remote', + title: 'Claude · deploy', + summary: 'approve `rm -rf build/`', + action: { type: 'open-session', sessionId: 'h-remote' }, + createdAt: '2026-07-06T12:00:00Z', +}; + const needsInput: AttentionItem = { id: 'needs-input:run-1', kind: 'needs-input', @@ -50,6 +63,21 @@ describe('AttentionQueueItem', () => { expect(onAction).toHaveBeenCalledWith({ type: 'approval', requestId: 'req-1' }); }); + it('renders a remote approval (approval kind) whose action opens its session', () => { + const onAction = vi.fn(); + render(); + const row = screen.getByTestId('attention-item-remote-approval:h-remote'); + expect(row).toHaveAttribute('data-kind', 'approval'); + // Approval kind → the amber "Review" verb, and the summary is shown. + expect(screen.getByText('Claude · deploy')).toBeInTheDocument(); + expect(screen.getByText('approve `rm -rf build/`')).toBeInTheDocument(); + const action = screen.getByTestId('attention-item-action'); + expect(action).toHaveTextContent('tinyplaceOrchestration.attention.review'); + // Clicking routes the open-session action so the tab opens the chat window. + fireEvent.click(action); + expect(onAction).toHaveBeenCalledWith({ type: 'open-session', sessionId: 'h-remote' }); + }); + it('renders a needs-input row with the Open action carrying its thread', () => { const onAction = vi.fn(); render(); diff --git a/app/src/components/intelligence/OrchestrationChatPrimitives.test.tsx b/app/src/components/intelligence/OrchestrationChatPrimitives.test.tsx index 7ba752e87..e43280610 100644 --- a/app/src/components/intelligence/OrchestrationChatPrimitives.test.tsx +++ b/app/src/components/intelligence/OrchestrationChatPrimitives.test.tsx @@ -74,4 +74,43 @@ describe('MessageBubble', () => { render(); expect(screen.getByText('••••')).toHaveClass('text-content-muted'); }); + + it('renders a tool_call as a monospace command row with ▶ + tool name', () => { + const { container } = render( + + ); + expect(container.querySelector('[data-event-kind="tool_call"]')).not.toBeNull(); + expect(screen.getByText('▶')).toBeInTheDocument(); + expect(screen.getByText('Bash')).toBeInTheDocument(); + expect(container.querySelector('p.font-mono')?.textContent).toBe('ls'); + }); + + it('renders a tool_result with the ↳ glyph', () => { + render( + + ); + expect(screen.getByText('↳')).toBeInTheDocument(); + }); + + it('renders agent_thinking italic + muted with the ∴ glyph', () => { + const { container } = render( + + ); + expect(container.querySelector('p.italic')).not.toBeNull(); + expect(screen.getByText('∴')).toBeInTheDocument(); + }); + + it('renders an error row with the ✕ glyph', () => { + render(); + expect(screen.getByText('✕')).toBeInTheDocument(); + }); + + it('falls back to the plain-dot style for a legacy v1 row (no eventKind)', () => { + const { container } = render(); + expect(container.querySelector('[data-event-kind="v1"]')).not.toBeNull(); + expect(container.querySelector('div.rounded-full')).not.toBeNull(); + expect(container.querySelector('p.font-mono')).toBeNull(); + }); }); diff --git a/app/src/components/intelligence/OrchestrationChatPrimitives.tsx b/app/src/components/intelligence/OrchestrationChatPrimitives.tsx index e7790231f..ea3e7270d 100644 --- a/app/src/components/intelligence/OrchestrationChatPrimitives.tsx +++ b/app/src/components/intelligence/OrchestrationChatPrimitives.tsx @@ -77,19 +77,97 @@ export function ChatListButton({ ); } +/** + * Per-kind presentation for a harness (v2) event row. Differentiates the typed + * stream (tool_call / tool_result / agent_thinking / approval_request / error) + * from plain agent/user messages so the orchestration thread reads like a live + * activity log rather than undifferentiated text. Legacy v1 rows (no + * `eventKind`) fall through to the default agent-message style. + */ +interface BubbleStyle { + /** Leading marker: a glyph when the kind has one, else a colored dot. */ + dot: string; + glyph?: string; + /** Render the body in a monospace block (tool command / output). */ + mono?: boolean; + /** Body text tone. */ + tone: string; + /** Optional left-accent on the bubble. */ + accent: string; +} + +function bubbleStyle(kind: ChatMessage['eventKind']): BubbleStyle { + switch (kind) { + case 'tool_call': + return { + dot: 'text-ocean-500', + glyph: '▶', + mono: true, + tone: 'text-content', + accent: 'border-l-2 border-l-ocean-400', + }; + case 'tool_result': + return { + dot: 'text-sage-500', + glyph: '↳', + mono: true, + tone: 'text-content-muted', + accent: 'border-l-2 border-l-sage-400', + }; + case 'agent_thinking': + return { + dot: 'text-content-faint', + glyph: '∴', + tone: 'italic text-content-faint', + accent: '', + }; + case 'approval_request': + return { + dot: 'text-amber-500', + glyph: '⚠', + tone: 'text-content', + accent: 'border-l-2 border-l-amber-400', + }; + case 'error': + return { + dot: 'text-coral-500', + glyph: '✕', + tone: 'text-coral-600 dark:text-coral-300', + accent: 'border-l-2 border-l-coral-400', + }; + case 'user_prompt': + return { dot: 'text-sage-500', tone: 'text-content', accent: '' }; + default: + return { dot: 'text-ocean-500', tone: 'text-content', accent: '' }; + } +} + export function MessageBubble({ message }: { message: ChatMessage }): ReactElement { + const style = bubbleStyle(message.eventKind); return ( -
-
-
+
+ {style.glyph ? ( + {style.glyph} + ) : ( +
+ )} +
{message.from} + {message.toolName ? ( + + {message.toolName} + + ) : null} {formatTime(message.timestamp)}

+ className={`mt-1 max-h-64 overflow-y-auto whitespace-pre-wrap break-words ${ + style.mono ? 'font-mono text-xs' : 'text-sm' + } ${message.encrypted ? 'text-content-muted' : style.tone}`}> {message.body}

diff --git a/app/src/components/intelligence/OrchestrationSidebar.test.tsx b/app/src/components/intelligence/OrchestrationSidebar.test.tsx index 364238412..8d09b9999 100644 --- a/app/src/components/intelligence/OrchestrationSidebar.test.tsx +++ b/app/src/components/intelligence/OrchestrationSidebar.test.tsx @@ -37,6 +37,9 @@ const props = (over: Partial): OrchestrationSidebarPr steeringText: null, selfIdentity: null, identityLoading: false, + onPublishIdentity: vi.fn(), + publishingIdentity: false, + publishIdentityError: null, attentionQueue: null, attentionLoading: false, onAttentionAction: vi.fn(), diff --git a/app/src/components/intelligence/OrchestrationSidebar.tsx b/app/src/components/intelligence/OrchestrationSidebar.tsx index 7dd34823f..9d19ca0d8 100644 --- a/app/src/components/intelligence/OrchestrationSidebar.tsx +++ b/app/src/components/intelligence/OrchestrationSidebar.tsx @@ -38,6 +38,9 @@ export interface OrchestrationSidebarProps { steeringText: string | null; selfIdentity: SelfIdentity | null; identityLoading: boolean; + onPublishIdentity: () => void; + publishingIdentity: boolean; + publishIdentityError: string | null; attentionQueue: AttentionQueue | null; attentionLoading: boolean; onAttentionAction: (action: AttentionAction) => void; @@ -72,6 +75,9 @@ export default function OrchestrationSidebar({ steeringText, selfIdentity, identityLoading, + onPublishIdentity, + publishingIdentity, + publishIdentityError, attentionQueue, attentionLoading, onAttentionAction, @@ -146,7 +152,13 @@ export default function OrchestrationSidebar({ ) : null}
- + { ).toBeInTheDocument(); }); + const undiscoverable: SelfIdentity = { + agentId: 'addrWithNoCardPublishedYet', + handles: [{ username: 'openhuman', primary: true }], + primaryHandle: 'openhuman', + cardPublished: false, + keyPublished: false, + discoverable: false, + }; + + it('renders a "Make discoverable" button that fires onPublish while undiscoverable', () => { + const onPublish = vi.fn(); + render(); + const btn = screen.getByTestId('tinyplace-self-identity-publish'); + expect(btn).toHaveTextContent('tinyplaceOrchestration.identity.makeDiscoverable'); + fireEvent.click(btn); + expect(onPublish).toHaveBeenCalledTimes(1); + }); + + it('disables the button and shows the publishing label while a publish is in flight', () => { + render( + + ); + const btn = screen.getByTestId('tinyplace-self-identity-publish'); + expect(btn).toBeDisabled(); + expect(btn).toHaveTextContent('tinyplaceOrchestration.identity.publishing'); + }); + + it('surfaces a publish error under the button', () => { + render( + + ); + expect(screen.getByTestId('tinyplace-self-identity-publish-error')).toHaveTextContent( + 'tinyplaceOrchestration.identity.publishFailed' + ); + }); + + it('shows no "make discoverable" button once discoverable', () => { + render(); + expect(screen.queryByTestId('tinyplace-self-identity-publish')).toBeNull(); + }); + + it('offers a "Republish keys" action while discoverable that fires onPublish', () => { + const onPublish = vi.fn(); + render(); + const btn = screen.getByTestId('tinyplace-self-identity-republish'); + expect(btn).toHaveTextContent('tinyplaceOrchestration.identity.republish'); + fireEvent.click(btn); + expect(onPublish).toHaveBeenCalledTimes(1); + }); + + it('omits the publish button when no onPublish handler is provided', () => { + render(); + expect(screen.queryByTestId('tinyplace-self-identity-publish')).toBeNull(); + }); + it('copies the address to the clipboard on click', async () => { render(); fireEvent.click(screen.getByTestId('tinyplace-self-identity-copy')); diff --git a/app/src/components/intelligence/SelfIdentityCard.tsx b/app/src/components/intelligence/SelfIdentityCard.tsx index 59ad8b69a..8a824d1a7 100644 --- a/app/src/components/intelligence/SelfIdentityCard.tsx +++ b/app/src/components/intelligence/SelfIdentityCard.tsx @@ -20,6 +20,16 @@ import type { SelfIdentity } from '../../lib/orchestration/orchestrationClient'; export interface SelfIdentityCardProps { identity: SelfIdentity | null; loading: boolean; + /** + * Publish (or refresh) this agent's directory card + Signal key so peers can + * DM it. Rendered only while undiscoverable. The parent owns the RPC + the + * identity refresh; the card just drives the button state. + */ + onPublish?: () => void; + /** True while {@link onPublish} is in flight — disables the button. */ + publishing?: boolean; + /** Non-null when the last publish attempt failed — surfaces under the button. */ + publishError?: string | null; } function shortAddress(address: string): string { @@ -30,6 +40,9 @@ function shortAddress(address: string): string { export default function SelfIdentityCard({ identity, loading, + onPublish, + publishing, + publishError, }: SelfIdentityCardProps): React.ReactElement { const { t } = useT(); const [copied, setCopied] = useState(false); @@ -124,10 +137,53 @@ export default function SelfIdentityCard({
+ {identity.discoverable && onPublish ? ( +
+ + {publishError ? ( + + {t('tinyplaceOrchestration.identity.publishFailed')} + + ) : null} +
+ ) : null} + {!identity.discoverable ? ( -

- {t('tinyplaceOrchestration.identity.undiscoverableHint')} -

+
+

+ {t('tinyplaceOrchestration.identity.undiscoverableHint')} +

+ {onPublish ? ( + + ) : null} + {publishError ? ( +

+ {t('tinyplaceOrchestration.identity.publishFailed')} +

+ ) : null} +
) : null} ); diff --git a/app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsx b/app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsx index 4f2edea1e..726b33d60 100644 --- a/app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsx +++ b/app/src/components/intelligence/TinyPlaceOrchestrationTab.test.tsx @@ -31,6 +31,7 @@ vi.mock('../../lib/orchestration/orchestrationClient', async importOriginal => { markRead: vi.fn(), status: vi.fn(), selfIdentity: vi.fn(), + publishIdentity: vi.fn(), relayInfo: vi.fn(), attention: vi.fn(), }, diff --git a/app/src/components/intelligence/TinyPlaceOrchestrationTab.tsx b/app/src/components/intelligence/TinyPlaceOrchestrationTab.tsx index f394a280e..20876685a 100644 --- a/app/src/components/intelligence/TinyPlaceOrchestrationTab.tsx +++ b/app/src/components/intelligence/TinyPlaceOrchestrationTab.tsx @@ -69,6 +69,10 @@ export default function TinyPlaceOrchestrationTab() { // the whole tab. const [selfIdentity, setSelfIdentity] = useState(null); const [identityLoading, setIdentityLoading] = useState(true); + // "Make discoverable" remediation: publishes the directory card + Signal key + // when the identity card reports the agent is un-messageable. + const [publishingIdentity, setPublishingIdentity] = useState(false); + const [publishIdentityError, setPublishIdentityError] = useState(null); const [relayInfo, setRelayInfo] = useState(null); // The aggregated "needs you" queue (approvals + blocked runs + unread). Read // independently of chats so a failure leaves the zone empty, never the tab. @@ -152,6 +156,28 @@ export default function TinyPlaceOrchestrationTab() { setIdentityLoading(false); }, []); + // Publish (or refresh) the directory card + Signal key, then adopt the fresh + // identity the RPC echoes back so the card flips to "discoverable" without a + // separate reload. A failure surfaces on the card, not the whole tab. + const publishIdentity = useCallback(async () => { + debug('[tinyplace-orchestration] publish identity entry'); + setPublishingIdentity(true); + setPublishIdentityError(null); + try { + const identity = await orchestrationClient.publishIdentity(); + if (!mountedRef.current) return; + debug('[tinyplace-orchestration] publish identity ok discoverable=%s', identity.discoverable); + setSelfIdentity(identity); + } catch (error) { + if (!mountedRef.current) return; + const message = error instanceof Error ? error.message : String(error); + debug('[tinyplace-orchestration] publish identity error %s', message); + setPublishIdentityError(message); + } finally { + if (mountedRef.current) setPublishingIdentity(false); + } + }, []); + const loadAttention = useCallback(async () => { debug('[tinyplace-orchestration] attention load entry'); try { @@ -348,6 +374,9 @@ export default function TinyPlaceOrchestrationTab() { steeringText={steeringText} selfIdentity={selfIdentity} identityLoading={identityLoading} + onPublishIdentity={publishIdentity} + publishingIdentity={publishingIdentity} + publishIdentityError={publishIdentityError} attentionQueue={attentionQueue} attentionLoading={attentionLoading} onAttentionAction={handleAttentionAction} diff --git a/app/src/lib/i18n/ar.ts b/app/src/lib/i18n/ar.ts index 4e2d3b7ae..be364a039 100644 --- a/app/src/lib/i18n/ar.ts +++ b/app/src/lib/i18n/ar.ts @@ -299,7 +299,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'قابل للاكتشاف', 'tinyplaceOrchestration.identity.undiscoverable': 'غير قابل للاكتشاف', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'سجّل @handle حتى يتمكن الآخرون من مراسلتك.', + 'انشر بطاقة الدليل ومفتاح التشفير حتى يتمكن الآخرون من مراسلتك.', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'اجعله قابلاً للاكتشاف', + 'tinyplaceOrchestration.identity.republish': 'إعادة نشر المفاتيح', + 'tinyplaceOrchestration.identity.publishing': 'جارٍ النشر…', + 'tinyplaceOrchestration.identity.publishFailed': 'فشل النشر — أعد المحاولة', 'tinyplaceOrchestration.identity.card': 'بطاقة الدليل', 'tinyplaceOrchestration.identity.key': 'مفتاح التشفير', 'tinyplaceOrchestration.identity.published': 'منشور', diff --git a/app/src/lib/i18n/bn.ts b/app/src/lib/i18n/bn.ts index a31f61442..d59283f50 100644 --- a/app/src/lib/i18n/bn.ts +++ b/app/src/lib/i18n/bn.ts @@ -306,7 +306,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'খুঁজে পাওয়া যায়', 'tinyplaceOrchestration.identity.undiscoverable': 'খুঁজে পাওয়া যায় না', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'একটি @handle নিবন্ধন করুন যাতে অন্য এজেন্টরা আপনাকে বার্তা পাঠাতে পারে।', + 'আপনার ডিরেক্টরি কার্ড ও এনক্রিপশন কী প্রকাশ করুন যাতে অন্য এজেন্টরা আপনাকে বার্তা পাঠাতে পারে।', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'আবিষ্কারযোগ্য করুন', + 'tinyplaceOrchestration.identity.republish': 'কী পুনঃপ্রকাশ করুন', + 'tinyplaceOrchestration.identity.publishing': 'প্রকাশ করা হচ্ছে…', + 'tinyplaceOrchestration.identity.publishFailed': 'প্রকাশ ব্যর্থ হয়েছে — আবার চেষ্টা করুন', 'tinyplaceOrchestration.identity.card': 'ডিরেক্টরি কার্ড', 'tinyplaceOrchestration.identity.key': 'এনক্রিপশন কী', 'tinyplaceOrchestration.identity.published': 'প্রকাশিত', diff --git a/app/src/lib/i18n/de.ts b/app/src/lib/i18n/de.ts index 13107b914..3f3d8cb6f 100644 --- a/app/src/lib/i18n/de.ts +++ b/app/src/lib/i18n/de.ts @@ -313,7 +313,12 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'Auffindbar', 'tinyplaceOrchestration.identity.undiscoverable': 'Nicht auffindbar', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'Registriere ein @handle, damit Peers dir schreiben können.', + 'Veröffentliche deine Verzeichniskarte und deinen Verschlüsselungsschlüssel, damit Peers dir schreiben können.', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'Auffindbar machen', + 'tinyplaceOrchestration.identity.republish': 'Schlüssel neu veröffentlichen', + 'tinyplaceOrchestration.identity.publishing': 'Wird veröffentlicht…', + 'tinyplaceOrchestration.identity.publishFailed': + 'Veröffentlichung fehlgeschlagen — erneut versuchen', 'tinyplaceOrchestration.identity.card': 'Verzeichniskarte', 'tinyplaceOrchestration.identity.key': 'Verschlüsselungsschlüssel', 'tinyplaceOrchestration.identity.published': 'Veröffentlicht', diff --git a/app/src/lib/i18n/en.ts b/app/src/lib/i18n/en.ts index 0210139d9..a805d7569 100644 --- a/app/src/lib/i18n/en.ts +++ b/app/src/lib/i18n/en.ts @@ -4248,7 +4248,11 @@ const en: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'Discoverable', 'tinyplaceOrchestration.identity.undiscoverable': 'Not discoverable', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'Register a @handle so peers can message you.', + 'Publish your directory card + encryption key so peers can message you.', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'Make discoverable', + 'tinyplaceOrchestration.identity.republish': 'Republish keys', + 'tinyplaceOrchestration.identity.publishing': 'Publishing…', + 'tinyplaceOrchestration.identity.publishFailed': 'Publish failed — try again', 'tinyplaceOrchestration.identity.card': 'Directory card', 'tinyplaceOrchestration.identity.key': 'Encryption key', 'tinyplaceOrchestration.identity.published': 'Published', diff --git a/app/src/lib/i18n/es.ts b/app/src/lib/i18n/es.ts index 6f77c60ab..2cf8e7ad5 100644 --- a/app/src/lib/i18n/es.ts +++ b/app/src/lib/i18n/es.ts @@ -310,7 +310,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'Detectable', 'tinyplaceOrchestration.identity.undiscoverable': 'No detectable', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'Registra un @handle para que otros agentes puedan escribirte.', + 'Publica tu tarjeta de directorio y tu clave de cifrado para que otros agentes puedan escribirte.', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'Hacer detectable', + 'tinyplaceOrchestration.identity.republish': 'Volver a publicar claves', + 'tinyplaceOrchestration.identity.publishing': 'Publicando…', + 'tinyplaceOrchestration.identity.publishFailed': 'Error al publicar — inténtalo de nuevo', 'tinyplaceOrchestration.identity.card': 'Tarjeta de directorio', 'tinyplaceOrchestration.identity.key': 'Clave de cifrado', 'tinyplaceOrchestration.identity.published': 'Publicado', diff --git a/app/src/lib/i18n/fr.ts b/app/src/lib/i18n/fr.ts index 457d9ed8c..87b5aae97 100644 --- a/app/src/lib/i18n/fr.ts +++ b/app/src/lib/i18n/fr.ts @@ -310,7 +310,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'Découvrable', 'tinyplaceOrchestration.identity.undiscoverable': 'Non découvrable', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'Enregistre un @handle pour que les pairs puissent te contacter.', + 'Publie ta fiche d’annuaire et ta clé de chiffrement pour que les pairs puissent te contacter.', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'Rendre découvrable', + 'tinyplaceOrchestration.identity.republish': 'Republier les clés', + 'tinyplaceOrchestration.identity.publishing': 'Publication…', + 'tinyplaceOrchestration.identity.publishFailed': 'Échec de la publication — réessaie', 'tinyplaceOrchestration.identity.card': 'Fiche d’annuaire', 'tinyplaceOrchestration.identity.key': 'Clé de chiffrement', 'tinyplaceOrchestration.identity.published': 'Publié', diff --git a/app/src/lib/i18n/hi.ts b/app/src/lib/i18n/hi.ts index 7a3973055..d829f0ae7 100644 --- a/app/src/lib/i18n/hi.ts +++ b/app/src/lib/i18n/hi.ts @@ -306,7 +306,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'खोजने योग्य', 'tinyplaceOrchestration.identity.undiscoverable': 'खोजने योग्य नहीं', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'एक @handle पंजीकृत करें ताकि अन्य एजेंट आपको संदेश भेज सकें।', + 'अपना डायरेक्टरी कार्ड और एन्क्रिप्शन कुंजी प्रकाशित करें ताकि अन्य एजेंट आपको संदेश भेज सकें।', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'खोजने योग्य बनाएँ', + 'tinyplaceOrchestration.identity.republish': 'कुंजियाँ पुनः प्रकाशित करें', + 'tinyplaceOrchestration.identity.publishing': 'प्रकाशित हो रहा है…', + 'tinyplaceOrchestration.identity.publishFailed': 'प्रकाशन विफल — पुनः प्रयास करें', 'tinyplaceOrchestration.identity.card': 'डायरेक्टरी कार्ड', 'tinyplaceOrchestration.identity.key': 'एन्क्रिप्शन कुंजी', 'tinyplaceOrchestration.identity.published': 'प्रकाशित', diff --git a/app/src/lib/i18n/id.ts b/app/src/lib/i18n/id.ts index 1a81cda28..ae5122164 100644 --- a/app/src/lib/i18n/id.ts +++ b/app/src/lib/i18n/id.ts @@ -307,7 +307,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'Dapat ditemukan', 'tinyplaceOrchestration.identity.undiscoverable': 'Tidak dapat ditemukan', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'Daftarkan @handle agar rekan dapat mengirimi Anda pesan.', + 'Terbitkan kartu direktori dan kunci enkripsi Anda agar rekan dapat mengirimi Anda pesan.', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'Jadikan dapat ditemukan', + 'tinyplaceOrchestration.identity.republish': 'Terbitkan ulang kunci', + 'tinyplaceOrchestration.identity.publishing': 'Menerbitkan…', + 'tinyplaceOrchestration.identity.publishFailed': 'Gagal menerbitkan — coba lagi', 'tinyplaceOrchestration.identity.card': 'Kartu direktori', 'tinyplaceOrchestration.identity.key': 'Kunci enkripsi', 'tinyplaceOrchestration.identity.published': 'Diterbitkan', diff --git a/app/src/lib/i18n/it.ts b/app/src/lib/i18n/it.ts index dbacdaee7..3e7d9f0f6 100644 --- a/app/src/lib/i18n/it.ts +++ b/app/src/lib/i18n/it.ts @@ -310,7 +310,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'Rilevabile', 'tinyplaceOrchestration.identity.undiscoverable': 'Non rilevabile', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'Registra un @handle così altri agenti possono scriverti.', + 'Pubblica la tua scheda directory e la tua chiave di crittografia così altri agenti possono scriverti.', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'Rendi rilevabile', + 'tinyplaceOrchestration.identity.republish': 'Ripubblica chiavi', + 'tinyplaceOrchestration.identity.publishing': 'Pubblicazione…', + 'tinyplaceOrchestration.identity.publishFailed': 'Pubblicazione non riuscita — riprova', 'tinyplaceOrchestration.identity.card': 'Scheda directory', 'tinyplaceOrchestration.identity.key': 'Chiave di crittografia', 'tinyplaceOrchestration.identity.published': 'Pubblicato', diff --git a/app/src/lib/i18n/ko.ts b/app/src/lib/i18n/ko.ts index 2ad9776fa..cdedbb1fc 100644 --- a/app/src/lib/i18n/ko.ts +++ b/app/src/lib/i18n/ko.ts @@ -303,7 +303,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': '검색 가능', 'tinyplaceOrchestration.identity.undiscoverable': '검색 불가', 'tinyplaceOrchestration.identity.undiscoverableHint': - '다른 에이전트가 메시지를 보낼 수 있도록 @handle을 등록하세요.', + '다른 에이전트가 메시지를 보낼 수 있도록 디렉터리 카드와 암호화 키를 게시하세요.', + 'tinyplaceOrchestration.identity.makeDiscoverable': '검색 가능하게 만들기', + 'tinyplaceOrchestration.identity.republish': '키 다시 게시', + 'tinyplaceOrchestration.identity.publishing': '게시 중…', + 'tinyplaceOrchestration.identity.publishFailed': '게시 실패 — 다시 시도하세요', 'tinyplaceOrchestration.identity.card': '디렉터리 카드', 'tinyplaceOrchestration.identity.key': '암호화 키', 'tinyplaceOrchestration.identity.published': '게시됨', diff --git a/app/src/lib/i18n/pl.ts b/app/src/lib/i18n/pl.ts index 25b13543f..f546e0572 100644 --- a/app/src/lib/i18n/pl.ts +++ b/app/src/lib/i18n/pl.ts @@ -311,7 +311,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'Wykrywalny', 'tinyplaceOrchestration.identity.undiscoverable': 'Niewykrywalny', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'Zarejestruj @handle, aby inni mogli do Ciebie pisać.', + 'Opublikuj swoją kartę katalogu i klucz szyfrowania, aby inni mogli do Ciebie pisać.', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'Ustaw jako wykrywalny', + 'tinyplaceOrchestration.identity.republish': 'Opublikuj ponownie klucze', + 'tinyplaceOrchestration.identity.publishing': 'Publikowanie…', + 'tinyplaceOrchestration.identity.publishFailed': 'Publikacja nie powiodła się — spróbuj ponownie', 'tinyplaceOrchestration.identity.card': 'Karta katalogu', 'tinyplaceOrchestration.identity.key': 'Klucz szyfrowania', 'tinyplaceOrchestration.identity.published': 'Opublikowano', diff --git a/app/src/lib/i18n/pt.ts b/app/src/lib/i18n/pt.ts index bd64f6a1c..a5c4a12aa 100644 --- a/app/src/lib/i18n/pt.ts +++ b/app/src/lib/i18n/pt.ts @@ -309,7 +309,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'Descobrível', 'tinyplaceOrchestration.identity.undiscoverable': 'Não descobrível', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'Registre um @handle para que outros agentes possam lhe enviar mensagens.', + 'Publique seu cartão de diretório e sua chave de criptografia para que outros agentes possam lhe enviar mensagens.', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'Tornar descobrível', + 'tinyplaceOrchestration.identity.republish': 'Republicar chaves', + 'tinyplaceOrchestration.identity.publishing': 'Publicando…', + 'tinyplaceOrchestration.identity.publishFailed': 'Falha ao publicar — tente novamente', 'tinyplaceOrchestration.identity.card': 'Cartão de diretório', 'tinyplaceOrchestration.identity.key': 'Chave de criptografia', 'tinyplaceOrchestration.identity.published': 'Publicado', diff --git a/app/src/lib/i18n/ru.ts b/app/src/lib/i18n/ru.ts index a6959c9ad..702823bf1 100644 --- a/app/src/lib/i18n/ru.ts +++ b/app/src/lib/i18n/ru.ts @@ -311,7 +311,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': 'Обнаруживаемый', 'tinyplaceOrchestration.identity.undiscoverable': 'Не обнаруживается', 'tinyplaceOrchestration.identity.undiscoverableHint': - 'Зарегистрируйте @handle, чтобы другие агенты могли писать вам.', + 'Опубликуйте карточку каталога и ключ шифрования, чтобы другие агенты могли писать вам.', + 'tinyplaceOrchestration.identity.makeDiscoverable': 'Сделать обнаруживаемым', + 'tinyplaceOrchestration.identity.republish': 'Опубликовать ключи заново', + 'tinyplaceOrchestration.identity.publishing': 'Публикация…', + 'tinyplaceOrchestration.identity.publishFailed': 'Не удалось опубликовать — попробуйте снова', 'tinyplaceOrchestration.identity.card': 'Карточка каталога', 'tinyplaceOrchestration.identity.key': 'Ключ шифрования', 'tinyplaceOrchestration.identity.published': 'Опубликовано', diff --git a/app/src/lib/i18n/zh-CN.ts b/app/src/lib/i18n/zh-CN.ts index 4eced2c31..37574dea4 100644 --- a/app/src/lib/i18n/zh-CN.ts +++ b/app/src/lib/i18n/zh-CN.ts @@ -287,7 +287,11 @@ const messages: TranslationMap = { 'tinyplaceOrchestration.identity.discoverable': '可被发现', 'tinyplaceOrchestration.identity.undiscoverable': '不可被发现', 'tinyplaceOrchestration.identity.undiscoverableHint': - '注册一个 @handle，以便其他代理可以给你发消息。', + '发布你的目录卡片和加密密钥，以便其他代理可以给你发消息。', + 'tinyplaceOrchestration.identity.makeDiscoverable': '设为可被发现', + 'tinyplaceOrchestration.identity.republish': '重新发布密钥', + 'tinyplaceOrchestration.identity.publishing': '正在发布…', + 'tinyplaceOrchestration.identity.publishFailed': '发布失败 — 请重试', 'tinyplaceOrchestration.identity.card': '目录卡片', 'tinyplaceOrchestration.identity.key': '加密密钥', 'tinyplaceOrchestration.identity.published': '已发布', diff --git a/app/src/lib/orchestration/orchestrationClient.ts b/app/src/lib/orchestration/orchestrationClient.ts index dc2efa8ec..27f7af6cd 100644 --- a/app/src/lib/orchestration/orchestrationClient.ts +++ b/app/src/lib/orchestration/orchestrationClient.ts @@ -50,6 +50,24 @@ export interface SessionSummary { pinned: boolean; } +/** + * Typed kind of a harness (v2) event, mirroring the SDK's + * `tinyplace.harness.session.v2` stream. Absent on legacy v1 rows (which carry + * only role + body). The core hides `status`/`lifecycle`/`unknown` from the + * thread, so those should not normally reach the renderer. + */ +export type HarnessEventKind = + | 'user_prompt' + | 'agent_message' + | 'agent_thinking' + | 'tool_call' + | 'tool_result' + | 'approval_request' + | 'status' + | 'lifecycle' + | 'error' + | 'unknown'; + export interface OrchestrationMessage { id: string; agentId: string; @@ -59,6 +77,12 @@ export interface OrchestrationMessage { body: string; timestamp: string; seq: number; + /** Typed event kind for v2 harness streams; absent on v1 text rows. */ + eventKind?: HarnessEventKind; + /** Raw harness tool name (e.g. "Bash") on tool_call / tool_result rows. */ + toolName?: string; + /** Correlation id shared by a tool_call and its tool_result. */ + callId?: string; } export interface OrchestrationSteering { @@ -258,6 +282,15 @@ export const orchestrationClient = { */ selfIdentity: () => call('openhuman.orchestration_self_identity', {}), + /** + * Make this agent discoverable: publish (or refresh) its directory card + Signal + * encryption key for the wallet's current identity, then return the updated + * {@link SelfIdentity}. No @handle registration and no payment — it repairs the + * common "has an identity but card/key aren't published" gap that makes every + * inbound DM 404. Powers the SelfIdentityCard's "Make discoverable" action. + */ + publishIdentity: () => call('openhuman.orchestration_publish_identity', {}), + /** The relay endpoint + network label the core is talking to (RelayBadge). */ relayInfo: () => call('openhuman.orchestration_relay_info', {}), }; diff --git a/app/src/lib/orchestration/useOrchestrationChats.ts b/app/src/lib/orchestration/useOrchestrationChats.ts index 4719450ef..d04d899c9 100644 --- a/app/src/lib/orchestration/useOrchestrationChats.ts +++ b/app/src/lib/orchestration/useOrchestrationChats.ts @@ -16,6 +16,7 @@ import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { socketService } from '../../services/socketService'; import { + type HarnessEventKind, orchestrationClient, type OrchestrationMessage, type OrchestrationMessageEvent, @@ -37,6 +38,12 @@ export interface ChatMessage { body: string; timestamp: string; encrypted: boolean; + /** Typed harness (v2) event kind; absent on legacy v1 text rows. */ + eventKind?: HarnessEventKind; + /** Tool name on tool_call / tool_result rows. */ + toolName?: string; + /** Correlation id linking a tool_call to its tool_result. */ + callId?: string; } export interface ChatWindow { @@ -84,6 +91,9 @@ function mapMessage(message: OrchestrationMessage): ChatMessage { body: message.body, timestamp: message.timestamp, encrypted: false, + ...(message.eventKind ? { eventKind: message.eventKind } : {}), + ...(message.toolName ? { toolName: message.toolName } : {}), + ...(message.callId ? { callId: message.callId } : {}), }; } diff --git a/src/openhuman/orchestration/attention.rs b/src/openhuman/orchestration/attention.rs index 1f830d370..78279ad5d 100644 --- a/src/openhuman/orchestration/attention.rs +++ b/src/openhuman/orchestration/attention.rs @@ -141,6 +141,25 @@ pub(crate) struct UnreadSignal { pub last_message_at: Option, } +/// A REMOTE agent session parked on a tool-approval decision. Phase 1 persists +/// `status.state = "waiting_approval"` on the session (plus the prompt in +/// `current_detail` and the in-flight `active_call_id`) when a v2 +/// `approval_request` event arrives from a peer harness. Unlike [`ApprovalSignal`] +/// — which is a call parked on THIS core's local approval gate — this is a +/// remote instance blocked on the user; acting on it opens the orchestration +/// chat window so the user can steer the reply. +pub(crate) struct RemoteApprovalSignal { + pub session_id: String, + /// Human-friendly session label; falls back to the session id at the builder. + pub label: Option, + /// The approval prompt / current-activity line (PII-safe summary text). + pub detail: Option, + /// The in-flight tool call id awaiting the decision, when known. Carried for + /// parity with the persisted run-state; not currently surfaced on the wire. + pub active_call_id: Option, + pub created_at: Option, +} + // ── Builders ──────────────────────────────────────────────────────────────── fn approval_item(sig: ApprovalSignal) -> AttentionItem { @@ -194,18 +213,44 @@ fn unread_item(sig: UnreadSignal) -> AttentionItem { } } +fn remote_approval_item(sig: RemoteApprovalSignal) -> AttentionItem { + let title = sig.label.unwrap_or_else(|| sig.session_id.clone()); + // Namespaced apart from local approvals (`approval:`) so a session + // id can never collide with a request id in the renderer's list keys, even + // though both surface as the same `Approval` kind. + AttentionItem { + id: format!("remote-approval:{}", sig.session_id), + kind: AttentionKind::Approval, + title, + summary: sig.detail, + count: None, + action: AttentionAction::OpenSession { + session_id: sig.session_id.clone(), + }, + instance_id: sig.session_id, + created_at: sig.created_at, + } +} + // ── Assembly ──────────────────────────────────────────────────────────────── -/// Fold the three signal sources into one ordered queue. Unread signals with a +/// Fold the four signal sources into one ordered queue. Unread signals with a /// zero count are dropped (nothing to surface). Ordering is by kind urgency, /// then newest-first within a kind, then id for a stable total order. +/// +/// `remote_approvals` are peer sessions parked on an approval decision (Phase 1's +/// persisted `waiting_approval` run-state). They share the `Approval` kind with +/// the local-gate `approvals`, so they lead the queue and are counted under +/// `counts.approvals` — they ARE approvals, just remote ones. pub(crate) fn assemble_attention( approvals: Vec, needs_input: Vec, unread: Vec, + remote_approvals: Vec, ) -> AttentionQueue { let mut items: Vec = Vec::new(); items.extend(approvals.into_iter().map(approval_item)); + items.extend(remote_approvals.into_iter().map(remote_approval_item)); items.extend(needs_input.into_iter().map(needs_input_item)); items.extend(unread.into_iter().filter(|s| s.unread > 0).map(unread_item)); @@ -281,6 +326,28 @@ pub(crate) fn needs_input_from_command_center( .unwrap_or_default() } +/// Map orchestration sessions into remote-approval signals — keeping only those +/// whose persisted v2 run-state is `waiting_approval` (Phase 1 stamps this when a +/// peer harness emits an `approval_request`). Pure: the handler passes the +/// already-fetched session list. `detail` carries the PII-safe approval prompt +/// (`current_detail`); the deep link opens the session window (`last_message_at` +/// stamps the newest-first tie-break). +pub(crate) fn remote_approval_signals( + sessions: Vec, +) -> Vec { + sessions + .into_iter() + .filter(|s| s.status_state.as_deref() == Some("waiting_approval")) + .map(|s| RemoteApprovalSignal { + session_id: s.session_id, + label: s.label, + detail: s.current_detail, + active_call_id: s.active_call_id, + created_at: Some(s.last_message_at), + }) + .collect() +} + #[cfg(test)] mod tests { use super::*; @@ -296,7 +363,7 @@ mod tests { #[test] fn empty_sources_yield_empty_queue_with_zero_counts() { - let q = assemble_attention(vec![], vec![], vec![]); + let q = assemble_attention(vec![], vec![], vec![], vec![]); assert!(q.items.is_empty()); assert_eq!( q.counts, @@ -326,6 +393,7 @@ mod tests { unread: 3, last_message_at: Some("2026-07-06T12:00:00Z".into()), }], + vec![], ); let kinds: Vec = q.items.iter().map(|i| i.kind).collect(); assert_eq!( @@ -357,6 +425,7 @@ mod tests { ], vec![], vec![], + vec![], ); let ids: Vec<&str> = q.items.iter().map(|i| i.instance_id.as_str()).collect(); assert_eq!(ids, vec!["new", "mid", "old"]); @@ -381,6 +450,7 @@ mod tests { last_message_at: Some("2026-07-06T12:00:00Z".into()), }, ], + vec![], ); assert_eq!(q.items.len(), 1); assert_eq!(q.items[0].instance_id, "loud"); @@ -416,6 +486,7 @@ mod tests { unread: 2, last_message_at: None, }], + vec![], ); let by_id = |id: &str| q.items.iter().find(|i| i.id == id).unwrap().clone(); @@ -505,6 +576,7 @@ mod tests { unread: 1, last_message_at: None, }], + vec![], ); assert_eq!(q.items[0].title, "h-xyz"); } @@ -613,4 +685,150 @@ mod tests { }])); assert!(signals.is_empty()); } + + // ── remote-approval mapping ───────────────────────────────────────────── + + use crate::openhuman::orchestration::types::OrchestrationSession; + + fn session(id: &str, state: Option<&str>) -> OrchestrationSession { + OrchestrationSession { + session_id: id.into(), + agent_id: "@peer".into(), + source: "claude".into(), + last_seq: 1, + created_at: "2026-07-06T00:00:00Z".into(), + last_message_at: "2026-07-06T12:00:00Z".into(), + status_state: state.map(str::to_string), + ..Default::default() + } + } + + #[test] + fn remote_approval_signals_keep_only_waiting_approval_sessions() { + let waiting = OrchestrationSession { + label: Some("Claude · deploy".into()), + current_detail: Some("approve `rm -rf build/`".into()), + active_call_id: Some("call-9".into()), + ..session("h-waiting", Some("waiting_approval")) + }; + let signals = remote_approval_signals(vec![ + waiting, + session("h-running", Some("running")), + session("h-idle", Some("idle")), + session("h-legacy", None), + ]); + assert_eq!(signals.len(), 1, "only the waiting_approval session maps"); + let s = &signals[0]; + assert_eq!(s.session_id, "h-waiting"); + assert_eq!(s.label.as_deref(), Some("Claude · deploy")); + assert_eq!(s.detail.as_deref(), Some("approve `rm -rf build/`")); + assert_eq!(s.active_call_id.as_deref(), Some("call-9")); + assert_eq!(s.created_at.as_deref(), Some("2026-07-06T12:00:00Z")); + } + + fn remote(session_id: &str, at: &str) -> RemoteApprovalSignal { + RemoteApprovalSignal { + session_id: session_id.into(), + label: Some("Claude · deploy".into()), + detail: Some("approve `rm -rf build/`".into()), + active_call_id: Some("call-9".into()), + created_at: Some(at.into()), + } + } + + #[test] + fn remote_approval_surfaces_as_approval_kind_ahead_of_other_kinds() { + let q = assemble_attention( + vec![], + vec![NeedsInputSignal { + run_id: "run-1".into(), + title: "researcher".into(), + summary: None, + thread_id: Some("t-1".into()), + updated_at: Some("2026-07-06T11:00:00Z".into()), + }], + vec![UnreadSignal { + session_id: "h-2".into(), + label: None, + unread: 2, + last_message_at: Some("2026-07-06T13:00:00Z".into()), + }], + vec![remote("h-remote", "2026-07-06T12:00:00Z")], + ); + // The remote approval leads (Approval kind, priority 0). + assert_eq!(q.items[0].kind, AttentionKind::Approval); + assert_eq!(q.items[0].id, "remote-approval:h-remote"); + assert_eq!( + q.items[0].action, + AttentionAction::OpenSession { + session_id: "h-remote".into() + } + ); + assert_eq!( + q.items[0].summary.as_deref(), + Some("approve `rm -rf build/`") + ); + assert_eq!(q.items[0].title, "Claude · deploy"); + // Counted under approvals — a remote approval IS an approval. + assert_eq!( + q.counts, + AttentionCounts { + total: 3, + approvals: 1, + needs_input: 1, + unread: 1 + } + ); + } + + #[test] + fn remote_approval_title_falls_back_to_session_id_when_unlabeled() { + let q = assemble_attention( + vec![], + vec![], + vec![], + vec![RemoteApprovalSignal { + session_id: "h-bare".into(), + label: None, + detail: None, + active_call_id: None, + created_at: None, + }], + ); + assert_eq!(q.items.len(), 1); + assert_eq!(q.items[0].title, "h-bare"); + assert_eq!(q.counts.approvals, 1); + } + + #[test] + fn empty_remote_approvals_is_a_no_op() { + // A queue with only a local approval is unchanged by an empty remote vec. + let q = assemble_attention( + vec![approval("r1", "2026-07-06T10:00:00Z")], + vec![], + vec![], + vec![], + ); + assert_eq!(q.items.len(), 1); + assert_eq!(q.items[0].id, "approval:r1"); + assert_eq!(q.counts.approvals, 1); + } + + #[test] + fn local_and_remote_approvals_coexist_newest_first() { + // Both are Approval kind; within the kind they order newest-first, so the + // later remote approval leads the earlier local one. + let q = assemble_attention( + vec![approval("local-old", "2026-07-06T09:00:00Z")], + vec![], + vec![], + vec![remote("remote-new", "2026-07-06T12:00:00Z")], + ); + assert_eq!(q.counts.approvals, 2); + let ids: Vec<&str> = q.items.iter().map(|i| i.id.as_str()).collect(); + assert_eq!( + ids, + vec!["remote-approval:remote-new", "approval:local-old"] + ); + } } diff --git a/src/openhuman/orchestration/ingest.rs b/src/openhuman/orchestration/ingest.rs index bb9f6c68d..2d096ae4f 100644 --- a/src/openhuman/orchestration/ingest.rs +++ b/src/openhuman/orchestration/ingest.rs @@ -14,7 +14,10 @@ use crate::openhuman::config::Config; use crate::openhuman::tinyplace::{acknowledge_message, decrypt_envelope}; use super::store; -use super::types::{ChatKind, OrchestrationMessage, OrchestrationSession, SessionEnvelopeV1}; +use super::types::{ + ChatKind, HarnessEventKind, OrchestrationMessage, OrchestrationSession, SessionEnvelopeV1, + SessionEnvelopeV2, +}; const LOG: &str = "orchestration"; @@ -71,9 +74,35 @@ struct ClassifiedMessage { source: String, label: Option, workspace: Option, + /// The wire `line` (v1) / `event.seq` (v2). Retained for parity/debugging only + /// — persistence deliberately IGNORES it and stamps a store ordinal (#4583). seq: i64, body: String, timestamp: String, + // ── v2 event shape (all `None`/false for v1 + Master) ──────────────────── + /// v2 `event.kind`; drives the per-message render branch (Phase 2). + event_kind: Option, + /// v2 tool identity for `tool_call` / `approval_request`. + tool_name: Option, + /// v2 `call_id` correlating `tool_result` → `tool_call`. + call_id: Option, + /// v2 `status.state` (or a derived state for approval/lifecycle/error) written + /// onto the session row so `derive_status` reads a real run-state. + status_state: Option, + /// v2 `status.detail` → the session's `current_detail` (roster task line). + status_detail: Option, + /// v2 `status.active_call_id` → the session's `active_call_id`. + active_call_id: Option, + /// Whether this event advances the monotonic wake ordinal. Content events + /// (prompts/messages/thinking/tool_call/tool_result/approval/error) do; pure + /// session-state events (`status`/`lifecycle`/`unknown`) do NOT, so a status + /// ping never spuriously wakes the front-end graph. + advances_seq: bool, + /// True only for an authoritative `status` snapshot, which OWNS the session's + /// run-state columns and must be able to CLEAR `current_detail`/`active_call_id` + /// (e.g. `running_tool` → `idle`). Content events leave this false so the + /// COALESCE upsert preserves the last status instead of wiping it. + authoritative_status: bool, } /// True for streams that carry ciphertext DM envelopes worth ingesting. @@ -101,49 +130,223 @@ fn order_prekey_bundles_first(messages: &mut [tinyplace::types::MessageEnvelope] }); } -/// Classify a decrypted DM: a harness envelope becomes a per-session message, -/// anything else becomes a message in the peer's Master window. Pure. +/// Classify a decrypted DM into the fields we persist. Version-dispatched: try a +/// v2 harness envelope first, then a v1 envelope, else the peer's Master window. +/// Both envelope versions discriminate on `envelope_version`, so the order is +/// safe (a v1 body never matches v2 and vice-versa); this is the v1↔v2 +/// coexistence seam — both persist into the same session model. Pure. fn classify_message(plaintext: String, fallback_timestamp: &str) -> ClassifiedMessage { - match SessionEnvelopeV1::parse(&plaintext) { - Some(env) => { - // Compute the session key while `env` is still fully intact (before any - // field moves below), since `session_key` borrows `&env`. - let session_id = env.session_key(); - let label = (env.scope.scope_type == "folder").then(|| env.scope.key.clone()); - let workspace = (!env.scope.cwd.is_empty()).then(|| env.scope.cwd.clone()); - let timestamp = if env.message.timestamp.is_empty() { - fallback_timestamp.to_string() - } else { - env.message.timestamp - }; - ClassifiedMessage { - chat_kind: ChatKind::Session, - // Key on the single per-pair session id (the shared `wrapper_session_id` - // both peers put on every message for a thread), so a reply threads back - // into the same session. Falls back to `harness_session_id` for a legacy - // envelope with no per-pair id. See `SessionEnvelopeV1::session_key`. - session_id, - role: env.message.role, - source: env.harness.provider, - label, - workspace, - seq: env.message.line, - body: env.message.text, - timestamp, - } - } - None => ClassifiedMessage { - chat_kind: ChatKind::Master, - session_id: "master".to_string(), - role: "user".to_string(), - source: String::new(), - label: None, - workspace: None, - seq: 0, - body: plaintext, - timestamp: fallback_timestamp.to_string(), - }, + if let Some(env) = SessionEnvelopeV2::parse(&plaintext) { + return classify_v2(env, fallback_timestamp); } + if let Some(env) = SessionEnvelopeV1::parse(&plaintext) { + return classify_v1(env, fallback_timestamp); + } + // Not a harness envelope → a plain DM in the peer's Master window. + ClassifiedMessage { + chat_kind: ChatKind::Master, + session_id: "master".to_string(), + role: "user".to_string(), + source: String::new(), + label: None, + workspace: None, + seq: 0, + body: plaintext, + timestamp: fallback_timestamp.to_string(), + event_kind: None, + tool_name: None, + call_id: None, + status_state: None, + status_detail: None, + active_call_id: None, + advances_seq: true, + authoritative_status: false, + } +} + +/// Classify a v1 harness envelope — the original per-session mapping, unchanged. +fn classify_v1(env: SessionEnvelopeV1, fallback_timestamp: &str) -> ClassifiedMessage { + // Compute the session key while `env` is still fully intact (before any + // field moves below), since `session_key` borrows `&env`. + let session_id = env.session_key(); + let label = (env.scope.scope_type == "folder").then(|| env.scope.key.clone()); + let workspace = (!env.scope.cwd.is_empty()).then(|| env.scope.cwd.clone()); + let timestamp = if env.message.timestamp.is_empty() { + fallback_timestamp.to_string() + } else { + env.message.timestamp + }; + ClassifiedMessage { + chat_kind: ChatKind::Session, + // Key on the single per-pair session id (the shared `wrapper_session_id` + // both peers put on every message for a thread), so a reply threads back + // into the same session. Falls back to `harness_session_id` for a legacy + // envelope with no per-pair id. See `SessionEnvelopeV1::session_key`. + session_id, + role: env.message.role, + source: env.harness.provider, + label, + workspace, + seq: env.message.line, + body: env.message.text, + timestamp, + event_kind: None, + tool_name: None, + call_id: None, + status_state: None, + status_detail: None, + active_call_id: None, + advances_seq: true, + authoritative_status: false, + } +} + +/// Classify a v2 harness envelope. Switches on `event.kind`, mapping each to the +/// persisted fields (per the plan's mapping table). Content events become thread +/// messages that advance the wake ordinal; `status`/`lifecycle`/`unknown` are +/// session-state-only (they still persist a row for id-dedupe + ack, but do NOT +/// advance the wake ordinal). Pure. +fn classify_v2(env: SessionEnvelopeV2, fallback_timestamp: &str) -> ClassifiedMessage { + let session_id = env.session_key(); + let label = (env.scope.scope_type == "folder").then(|| env.scope.key.clone()); + let workspace = (!env.scope.cwd.is_empty()).then(|| env.scope.cwd.clone()); + let source = env.harness.provider.clone(); + let timestamp = if env.event.ts.is_empty() { + fallback_timestamp.to_string() + } else { + env.event.ts.clone() + }; + let kind_str = env.event.kind.clone(); + let wire_role = env.event.role.clone(); + let seq = env.event.seq; + + // Per-kind body + tool/status fields + wake disposition. + let mut b = V2Body::default(); + match env.event.decoded() { + HarnessEventKind::UserPrompt(p) => { + b.body = p.text; + b.default_role = "owner"; + } + HarnessEventKind::AgentMessage(p) => { + b.body = p.text; + } + HarnessEventKind::AgentThinking(p) => { + b.body = p.text; + } + HarnessEventKind::ToolCall(p) => { + b.body = p.display; + b.tool_name = non_empty(p.tool_name); + b.call_id = non_empty(p.call_id); + } + HarnessEventKind::ToolResult(p) => { + b.body = p.output; + b.call_id = non_empty(p.call_id); + } + HarnessEventKind::ApprovalRequest(p) => { + b.body = p.display; + b.tool_name = non_empty(p.tool_name); + b.call_id = p.call_id.and_then(non_empty); + // Drive the roster dot to waiting-approval. + b.status_state = Some("waiting_approval".to_string()); + } + HarnessEventKind::Status(p) => { + b.body = p.detail.clone(); + b.status_state = non_empty(p.state); + b.status_detail = non_empty(p.detail); + b.active_call_id = p.active_call_id.and_then(non_empty); + b.advances_seq = false; + // Authoritative run-state snapshot: it may CLEAR detail/active_call_id + // (running_tool → idle), so persistence overwrites rather than COALESCEs. + b.authoritative_status = true; + } + HarnessEventKind::Lifecycle(p) => { + b.body = p.phase.clone(); + // A session_end lifecycle marks the instance stopped; other phases + // carry no run-state. + if p.phase == "session_end" { + b.status_state = Some("stopped".to_string()); + } + b.advances_seq = false; + } + HarnessEventKind::Error(p) => { + b.body = p.message; + b.status_state = Some("errored".to_string()); + } + HarnessEventKind::Unknown(p) => { + // Preserve the raw payload as the body so nothing is silently lost. + b.body = serde_json::to_string(&p.raw).unwrap_or_default(); + b.advances_seq = false; + // Persist as the literal `unknown` (not the raw wire kind) so the store + // readers keep a forward/garbled event out of the thread + unread count. + b.kind_override = Some("unknown"); + } + } + + let role = if !wire_role.is_empty() { + wire_role + } else { + b.default_role.to_string() + }; + + ClassifiedMessage { + chat_kind: ChatKind::Session, + session_id, + role, + source, + label, + workspace, + seq, + body: b.body, + timestamp, + event_kind: Some(b.kind_override.map(str::to_string).unwrap_or(kind_str)), + tool_name: b.tool_name, + call_id: b.call_id, + status_state: b.status_state, + status_detail: b.status_detail, + active_call_id: b.active_call_id, + advances_seq: b.advances_seq, + authoritative_status: b.authoritative_status, + } +} + +/// Per-kind accumulator for [`classify_v2`], so the big match stays a set of small +/// assignments with sensible defaults (content event, `agent` role). +struct V2Body { + body: String, + default_role: &'static str, + tool_name: Option, + call_id: Option, + status_state: Option, + status_detail: Option, + active_call_id: Option, + advances_seq: bool, + authoritative_status: bool, + /// Overrides the persisted `event_kind` for a forward/garbled kind that + /// `decoded()` folded to `Unknown`: stored as the literal `"unknown"` so the + /// store readers keep it out of the thread instead of leaking the raw kind. + kind_override: Option<&'static str>, +} + +impl Default for V2Body { + fn default() -> Self { + V2Body { + body: String::new(), + default_role: "agent", + tool_name: None, + call_id: None, + status_state: None, + status_detail: None, + active_call_id: None, + advances_seq: true, + authoritative_status: false, + kind_override: None, + } + } +} + +/// `Some(s)` when non-empty, else `None` — so blank wire strings persist as NULL. +fn non_empty(s: String) -> Option { + (!s.is_empty()).then_some(s) } /// Persist a classified message + its session row. Idempotent by `msg_id`; @@ -173,7 +376,17 @@ fn persist_message( // concurrent writer on the same session (the drain here vs the graph's // `send_dm` reply persist) can't read the same `MAX(seq)` and duplicate it. store::in_immediate_txn(c, |c| { - let ingest_seq = store::next_session_seq(c, agent_id, &classified.session_id)?; + // Content events advance the monotonic wake ordinal (the #4583 fix); + // pure session-state events (status/lifecycle/unknown) persist a row + // for id-dedupe + ack but stamp `seq = 0` so they do NOT advance + // `last_seq` and therefore never spuriously wake the front-end graph. + // (`upsert_session` clamps `last_seq` with `MAX(...)`, so a 0 here only + // refreshes `last_message_at` + the status columns.) + let ingest_seq = if classified.advances_seq { + store::next_session_seq(c, agent_id, &classified.session_id)? + } else { + 0 + }; store::upsert_session( c, &OrchestrationSession { @@ -185,9 +398,25 @@ fn persist_message( last_seq: ingest_seq, created_at: now.to_string(), last_message_at: classified.timestamp.clone(), + status_state: classified.status_state.clone(), + current_detail: classified.status_detail.clone(), + active_call_id: classified.active_call_id.clone(), }, )?; - store::insert_message( + // An authoritative `status` snapshot OWNS the run-state columns and may + // CLEAR them (running_tool → idle); the COALESCE upsert above can't, so + // overwrite them directly. Content events skip this and keep COALESCE. + if classified.authoritative_status { + store::apply_run_state( + c, + agent_id, + &classified.session_id, + classified.status_state.as_deref(), + classified.status_detail.as_deref(), + classified.active_call_id.as_deref(), + )?; + } + let landed = store::insert_message( c, &OrchestrationMessage { id: msg_id.to_string(), @@ -198,8 +427,18 @@ fn persist_message( body: classified.body.clone(), timestamp: classified.timestamp.clone(), seq: ingest_seq, + event_kind: classified.event_kind.clone(), + tool_name: classified.tool_name.clone(), + call_id: classified.call_id.clone(), }, - ) + )?; + // An `error` event also records the short cause on the status surface + // (`orchestration.status.last_error`). Body is a harness error message, + // never a response body — safe to store (workspace-internal DB). + if landed && classified.event_kind.as_deref() == Some("error") { + store::kv_set(c, "orchestration:last_error", &classified.body)?; + } + Ok(landed) }) }) .map_err(|e| format!("persist: {e}")) @@ -461,6 +700,329 @@ mod tests { assert_eq!(c.timestamp, "2026-07-02T09:00:00Z"); // fallback used } + /// A v2 wire envelope for `kind`/`payload`, keyed on `wrapper` session id. + fn v2_env(kind: &str, payload: &str, wrapper: &str, role: &str) -> String { + format!( + r#"{{ + "envelope_version": "tinyplace.harness.session.v2", + "version": 2, + "scope": {{ "type": "folder", "key": "my-repo", "cwd": "/w", + "wrapper_session_id": "{wrapper}", "harness_session_id": "h2" }}, + "harness": {{ "provider": "claude", "command": "claude", "argv": [] }}, + "event": {{ "id": "e1", "seq": 9, "ts": "2026-07-05T01:00:00Z", + "role": "{role}", "kind": "{kind}", "payload": {payload} }}, + "source": {{ "path": "p", "record_type": "assistant" }} + }}"# + ) + } + + #[test] + fn classifies_v2_content_events_with_kind_and_tool_fields() { + // agent_message → session message, role from wire, event_kind set. + let am = classify_message( + v2_env("agent_message", r#"{ "text": "on it" }"#, "w2", "agent"), + "2026-07-05T09:00:00Z", + ); + assert_eq!(am.chat_kind, ChatKind::Session); + assert_eq!(am.session_id, "w2"); // shared wrapper id, same as v1 + assert_eq!(am.source, "claude"); + assert_eq!(am.label.as_deref(), Some("my-repo")); + assert_eq!(am.role, "agent"); + assert_eq!(am.body, "on it"); + assert_eq!(am.event_kind.as_deref(), Some("agent_message")); + assert_eq!(am.timestamp, "2026-07-05T01:00:00Z"); // event ts wins + assert!(am.advances_seq); + + // user_prompt defaults role to owner when the wire role is blank. + let up = classify_message( + v2_env( + "user_prompt", + r#"{ "text": "go", "source": "human" }"#, + "w2", + "", + ), + "2026-07-05T09:00:00Z", + ); + assert_eq!(up.role, "owner"); + assert_eq!(up.body, "go"); + + // tool_call → body is the display, tool_name + call_id captured. + let tc = classify_message( + v2_env( + "tool_call", + r#"{ "call_id": "c1", "tool_name": "bash", "tool_kind": "shell", "display": "ls" }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:00Z", + ); + assert_eq!(tc.event_kind.as_deref(), Some("tool_call")); + assert_eq!(tc.body, "ls"); + assert_eq!(tc.tool_name.as_deref(), Some("bash")); + assert_eq!(tc.call_id.as_deref(), Some("c1")); + assert!(tc.advances_seq); + + // tool_result → body is the output, call_id correlates back to the call. + let tr = classify_message( + v2_env( + "tool_result", + r#"{ "call_id": "c1", "ok": true, "output": "file.rs", "output_bytes": 7 }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:00Z", + ); + assert_eq!(tr.event_kind.as_deref(), Some("tool_result")); + assert_eq!(tr.body, "file.rs"); + assert_eq!(tr.call_id.as_deref(), Some("c1")); + } + + #[test] + fn classifies_v2_status_and_approval_and_error_run_state() { + // status → session-state-only: sets status_state/detail, does NOT advance seq. + let st = classify_message( + v2_env( + "status", + r#"{ "state": "running_tool", "detail": "compiling", "active_call_id": "c1" }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:00Z", + ); + assert_eq!(st.event_kind.as_deref(), Some("status")); + assert_eq!(st.status_state.as_deref(), Some("running_tool")); + assert_eq!(st.status_detail.as_deref(), Some("compiling")); + assert_eq!(st.active_call_id.as_deref(), Some("c1")); + assert!(!st.advances_seq, "status must not advance the wake ordinal"); + assert!( + st.authoritative_status, + "a status snapshot owns the run-state columns (may clear them)" + ); + + // approval_request → drives waiting_approval, keeps advancing (owner must act). + let ar = classify_message( + v2_env( + "approval_request", + r#"{ "call_id": "c9", "tool_name": "rm", "display": "rm -rf x" }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:00Z", + ); + assert_eq!(ar.status_state.as_deref(), Some("waiting_approval")); + assert_eq!(ar.tool_name.as_deref(), Some("rm")); + assert!(ar.advances_seq); + assert!( + !ar.authoritative_status, + "approval sets status_state but is not a run-state snapshot (COALESCE)" + ); + + // error → errored run-state + body is the message. + let er = classify_message( + v2_env( + "error", + r#"{ "message": "boom", "fatal": true }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:00Z", + ); + assert_eq!(er.status_state.as_deref(), Some("errored")); + assert_eq!(er.body, "boom"); + + // lifecycle session_end → stopped, session-state-only. + let lc = classify_message( + v2_env("lifecycle", r#"{ "phase": "session_end" }"#, "w2", "agent"), + "2026-07-05T09:00:00Z", + ); + assert_eq!(lc.status_state.as_deref(), Some("stopped")); + assert!(!lc.advances_seq); + + // unknown → dropped from the wake path, raw preserved as body. The raw + // wire kind ("teleport") is normalized to the literal "unknown" so store + // readers keep it out of the thread instead of leaking a forward kind. + let uk = classify_message( + v2_env("teleport", r#"{ "flux": 1 }"#, "w2", "agent"), + "2026-07-05T09:00:00Z", + ); + assert_eq!(uk.event_kind.as_deref(), Some("unknown")); + assert!(!uk.advances_seq); + assert!(uk.body.contains("flux")); + } + + #[test] + fn v1_and_v2_envelopes_coexist_in_one_session_model() { + // During rollout a wrapped harness may emit v1 then v2 under the SAME + // shared wrapper id. Both classify to ChatKind::Session under the same + // session_id and persist into one session — proving the coexistence seam. + let tmp = tempfile::tempdir().unwrap(); + let v1 = classify_message(ENVELOPE.to_string(), "2026-07-02T09:00:00Z"); // wrapper w1 + let v2 = classify_message( + v2_env("agent_message", r#"{ "text": "v2 line" }"#, "w1", "agent"), + "2026-07-05T09:00:00Z", + ); + assert_eq!(v1.session_id, "w1"); + assert_eq!(v2.session_id, "w1"); + + assert!(persist_message(tmp.path(), "m-v1", "@peer", &v1, "now").unwrap()); + assert!(persist_message(tmp.path(), "m-v2", "@peer", &v2, "now").unwrap()); + store::with_connection(tmp.path(), |c| { + // Both rows land in the single "w1" session, seqs monotonic 1,2. + assert_eq!(store::count_messages(c, "@peer", "w1")?, 2); + let seqs: Vec = store::list_recent_messages(c, "@peer", "w1", 10)? + .iter() + .map(|m| m.seq) + .collect(); + assert_eq!(seqs, vec![1, 2]); + Ok(()) + }) + .unwrap(); + } + + #[test] + fn persisting_v2_status_updates_session_row_without_advancing_seq() { + let tmp = tempfile::tempdir().unwrap(); + // A content message establishes the session at seq 1. + let msg = classify_message( + v2_env("agent_message", r#"{ "text": "working" }"#, "w2", "agent"), + "2026-07-05T09:00:00Z", + ); + assert!(persist_message(tmp.path(), "m1", "@peer", &msg, "now").unwrap()); + // A status event follows: it lands a (deduped) row + updates the session + // status columns, but last_seq must stay at 1 (no spurious wake). + let status = classify_message( + v2_env( + "status", + r#"{ "state": "running", "detail": "still working", "active_call_id": "c3" }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:00Z", + ); + assert!(persist_message(tmp.path(), "m2", "@peer", &status, "now").unwrap()); + + store::with_connection(tmp.path(), |c| { + let session = store::load_session(c, "@peer", "w2")?.expect("session exists"); + assert_eq!(session.status_state.as_deref(), Some("running")); + assert_eq!(session.current_detail.as_deref(), Some("still working")); + assert_eq!(session.active_call_id.as_deref(), Some("c3")); + assert_eq!(session.last_seq, 1, "status must not advance last_seq"); + // The status row is persisted (id-dedupe) with its event_kind + seq 0. + let msgs = store::list_recent_messages(c, "@peer", "w2", 10)?; + let status_row = msgs + .iter() + .find(|m| m.event_kind.as_deref() == Some("status")) + .expect("status row persisted"); + assert_eq!(status_row.seq, 0); + Ok(()) + }) + .unwrap(); + } + + #[test] + fn a_later_status_snapshot_clears_stale_detail_and_active_call_id() { + let tmp = tempfile::tempdir().unwrap(); + // running_tool: detail + active_call_id populated. + let running = classify_message( + v2_env( + "status", + r#"{ "state": "running_tool", "detail": "compiling", "active_call_id": "c1" }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:00Z", + ); + assert!(persist_message(tmp.path(), "s1", "@peer", &running, "now").unwrap()); + store::with_connection(tmp.path(), |c| { + let s = store::load_session(c, "@peer", "w2")?.expect("session"); + assert_eq!(s.current_detail.as_deref(), Some("compiling")); + assert_eq!(s.active_call_id.as_deref(), Some("c1")); + Ok(()) + }) + .unwrap(); + + // idle: the harness went quiet — detail + active_call_id are absent. The + // authoritative snapshot must CLEAR them, not keep the stale "compiling"/c1 + // (the COALESCE upsert alone would preserve them — the bug this fixes). + let idle = classify_message( + v2_env("status", r#"{ "state": "idle" }"#, "w2", "agent"), + "2026-07-05T09:01:00Z", + ); + assert!(persist_message(tmp.path(), "s2", "@peer", &idle, "now").unwrap()); + store::with_connection(tmp.path(), |c| { + let s = store::load_session(c, "@peer", "w2")?.expect("session"); + assert_eq!(s.status_state.as_deref(), Some("idle")); + assert_eq!(s.current_detail, None, "stale detail must be cleared"); + assert_eq!( + s.active_call_id, None, + "stale active_call_id must be cleared" + ); + Ok(()) + }) + .unwrap(); + } + + #[test] + fn a_content_event_never_wipes_a_live_status() { + let tmp = tempfile::tempdir().unwrap(); + // Establish a live run-state via a status snapshot. + let running = classify_message( + v2_env( + "status", + r#"{ "state": "running_tool", "detail": "compiling", "active_call_id": "c1" }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:00Z", + ); + assert!(persist_message(tmp.path(), "s1", "@peer", &running, "now").unwrap()); + // A content event (agent_message) carries no run-state — it must COALESCE, + // preserving the live status rather than nulling it. + let msg = classify_message( + v2_env( + "agent_message", + r#"{ "text": "still going" }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:30Z", + ); + assert!(persist_message(tmp.path(), "m1", "@peer", &msg, "now").unwrap()); + store::with_connection(tmp.path(), |c| { + let s = store::load_session(c, "@peer", "w2")?.expect("session"); + assert_eq!(s.status_state.as_deref(), Some("running_tool")); + assert_eq!(s.current_detail.as_deref(), Some("compiling")); + assert_eq!(s.active_call_id.as_deref(), Some("c1")); + Ok(()) + }) + .unwrap(); + } + + #[test] + fn persisting_v2_error_records_last_error() { + let tmp = tempfile::tempdir().unwrap(); + let err = classify_message( + v2_env( + "error", + r#"{ "message": "rate limited", "fatal": false }"#, + "w2", + "agent", + ), + "2026-07-05T09:00:00Z", + ); + assert!(persist_message(tmp.path(), "e1", "@peer", &err, "now").unwrap()); + store::with_connection(tmp.path(), |c| { + assert_eq!( + store::kv_get(c, "orchestration:last_error")?.as_deref(), + Some("rate limited") + ); + let session = store::load_session(c, "@peer", "w2")?.unwrap(); + assert_eq!(session.status_state.as_deref(), Some("errored")); + Ok(()) + }) + .unwrap(); + } + #[test] fn persist_message_is_idempotent_and_buckets_by_session() { let tmp = tempfile::tempdir().unwrap(); diff --git a/src/openhuman/orchestration/ops.rs b/src/openhuman/orchestration/ops.rs index 8bbd02da5..78e23ad01 100644 --- a/src/openhuman/orchestration/ops.rs +++ b/src/openhuman/orchestration/ops.rs @@ -348,6 +348,7 @@ fn thread_reply_to_origin( last_seq: seq, created_at: now.clone(), last_message_at: now.clone(), + ..Default::default() }, )?; store::insert_message( @@ -361,6 +362,7 @@ fn thread_reply_to_origin( body, timestamp: now.clone(), seq, + ..Default::default() }, ) }); @@ -445,6 +447,7 @@ async fn report_peer_reply_to_master( last_seq: seq, created_at: now.clone(), last_message_at: now.clone(), + ..Default::default() }, )?; store::insert_message( @@ -458,6 +461,7 @@ async fn report_peer_reply_to_master( body: report.to_string(), timestamp: now.clone(), seq, + ..Default::default() }, ) }) @@ -691,6 +695,7 @@ pub async fn record_subconscious_directive(config: &Config, directive_id: i64, t last_seq: directive_id, created_at: now.clone(), last_message_at: now.clone(), + ..Default::default() }, )?; store::insert_message( @@ -704,6 +709,7 @@ pub async fn record_subconscious_directive(config: &Config, directive_id: i64, t body: text.to_string(), timestamp: now.clone(), seq: directive_id, + ..Default::default() }, ) }) { @@ -941,6 +947,7 @@ impl ProductionRuntime { last_seq: 0, created_at: now.clone(), last_message_at: now.clone(), + ..Default::default() }, )?; store::insert_message( @@ -954,6 +961,7 @@ impl ProductionRuntime { body: body.to_string(), timestamp: now.clone(), seq, + ..Default::default() }, ) }) @@ -1274,6 +1282,7 @@ impl OrchestrationRuntime for ProductionRuntime { last_seq: seq, created_at: now.clone(), last_message_at: now.clone(), + ..Default::default() }, )?; store::insert_message( @@ -1287,6 +1296,7 @@ impl OrchestrationRuntime for ProductionRuntime { body: body_owned, timestamp: now.clone(), seq, + ..Default::default() }, ) }) @@ -1486,6 +1496,27 @@ pub(super) fn gather_unread_signals( Ok(out) } +/// Gather remote-approval attention signals from the orchestration store: every +/// session whose persisted v2 run-state is `waiting_approval` (Phase 1 stamps +/// this — plus the prompt in `current_detail` and the in-flight `active_call_id` +/// — when a peer harness emits an `approval_request`). Mirrors +/// [`gather_unread_signals`]: pure store read, the pure mapper +/// [`super::attention::remote_approval_signals`] does the filtering. +pub(super) fn gather_remote_approval_signals( + conn: &rusqlite::Connection, +) -> anyhow::Result> { + let signals = super::attention::remote_approval_signals(store::list_sessions(conn)?); + for sig in &signals { + log::debug!( + target: LOG, + "[orchestration_rpc] attention.remote_approval session_id={} has_call={}", + sig.session_id, + sig.active_call_id.is_some(), + ); + } + Ok(signals) +} + #[cfg(test)] mod tests { use super::*; @@ -1510,6 +1541,7 @@ mod tests { last_seq: 1, created_at: "2026-07-06T00:00:00Z".into(), last_message_at: at.into(), + ..Default::default() }; let message = |id: &str, session: &str, kind: ChatKind, at: &str| OrchestrationMessage { id: id.into(), @@ -1520,6 +1552,7 @@ mod tests { body: "hello".into(), timestamp: at.into(), seq: 1, + ..Default::default() }; let signals = store::with_connection(&config.workspace_dir, |conn| { @@ -1660,6 +1693,7 @@ mod tests { body: "hello".into(), timestamp: format!("2026-07-02T00:00:{seq:02}Z"), seq, + ..Default::default() } } @@ -2021,6 +2055,7 @@ mod tests { last_seq: 1, created_at: "now".into(), last_message_at: "now".into(), + ..Default::default() }, )?; store::insert_message(conn, &msg("h1", 1))?; diff --git a/src/openhuman/orchestration/schemas.rs b/src/openhuman/orchestration/schemas.rs index 6619af242..5fab2b74e 100644 --- a/src/openhuman/orchestration/schemas.rs +++ b/src/openhuman/orchestration/schemas.rs @@ -33,6 +33,7 @@ pub fn all_controller_schemas() -> Vec { schema_for("orchestration_status"), schema_for("orchestration_attention"), schema_for("orchestration_self_identity"), + schema_for("orchestration_publish_identity"), schema_for("orchestration_relay_info"), ] } @@ -71,6 +72,10 @@ pub fn all_registered_controllers() -> Vec { schema: schema_for("orchestration_self_identity"), handler: handle_self_identity, }, + RegisteredController { + schema: schema_for("orchestration_publish_identity"), + handler: handle_publish_identity, + }, RegisteredController { schema: schema_for("orchestration_relay_info"), handler: handle_relay_info, @@ -141,7 +146,7 @@ fn schema_for(function: &str) -> ControllerSchema { "orchestration_attention" => ControllerSchema { namespace: "orchestration", function: "attention", - description: "Aggregate the \"needs you\" signals across the hub — pending tool approvals, agent runs awaiting input, and instances with unread messages — into one priority-ordered queue.", + description: "Aggregate the \"needs you\" signals across the hub — pending local tool approvals, remote sessions parked on an approval, agent runs awaiting input, and instances with unread messages — into one priority-ordered queue.", inputs: vec![], outputs: vec![json_output("result", "AttentionQueue { items: AttentionItem[], counts }.")], }, @@ -155,6 +160,16 @@ fn schema_for(function: &str) -> ControllerSchema { "{ agentId, handles: {username, primary}[], primaryHandle?, cardPublished, keyPublished, discoverable }.", )], }, + "orchestration_publish_identity" => ControllerSchema { + namespace: "orchestration", + function: "publish_identity", + description: "Make this agent discoverable: publish (or refresh) its directory card and Signal encryption key for the current wallet identity, then return the updated self-identity. No @handle registration or payment — repairs the common \"has identity but card/key unpublished\" gap that makes inbound DMs 404. Returns the same shape as self_identity.", + inputs: vec![], + outputs: vec![json_output( + "result", + "{ agentId, handles: {username, primary}[], primaryHandle?, cardPublished, keyPublished, discoverable }.", + )], + }, "orchestration_relay_info" => ControllerSchema { namespace: "orchestration", function: "relay_info", @@ -262,17 +277,40 @@ fn harness_type_for(source: &str) -> Option { matches!(source, "claude" | "codex" | "gemini").then(|| source.to_string()) } -/// Coarse instance status for the roster dot, derived from activity. Peer -/// instances carry no true run-state yet, so today an instance is `idle` when it -/// has recent traffic and `stopped` otherwise. The richer -/// running/waiting-approval/errored states are reserved for the attention-queue -/// and run-state follow-ups; the renderer's `InstanceStatusDot` already models -/// all five. -fn derive_status(active: bool) -> &'static str { - if active { - "idle" - } else { - "stopped" +/// Coarse instance status for the roster dot. Reads the persisted v2 run-state +/// (`status.state`) when present, mapping it to the renderer's five +/// `InstanceStatus` values; falls back to recency for v1/legacy sessions that +/// carry no run-state. +/// +/// Staleness fallback: a `running`/`running_tool` session that has gone silent +/// (no recent traffic → `active == false`) is reported `stopped`, so a crashed +/// instance that never emitted a terminal `status` doesn't sit forever on a stuck +/// green dot. `waiting_approval`/`errored`/`stopped`/`idle` are honoured as-is — +/// they are already terminal/blocked states, not "silently alive" ones. +fn derive_status(status_state: Option<&str>, active: bool) -> &'static str { + match status_state { + // Actively working — but downgrade to stopped if the session has gone + // silent (staleness fallback). + Some("running") | Some("running_tool") => { + if active { + "running" + } else { + "stopped" + } + } + Some("waiting_approval") => "waiting-approval", + Some("idle") => "idle", + Some("stopped") => "stopped", + Some("errored") => "errored", + // No persisted run-state (v1/legacy) or an unrecognised value → the + // original recency heuristic. + _ => { + if active { + "idle" + } else { + "stopped" + } + } } } @@ -299,7 +337,7 @@ fn summarize( let chat_kind = chat_kind_for_session(&session.session_id); let active = pinned || is_active(&session.last_message_at); let harness_type = harness_type_for(&session.source); - let status = derive_status(active).to_string(); + let status = derive_status(session.status_state.as_deref(), active).to_string(); SessionSummary { chat_kind: chat_kind.as_str().to_string(), active, @@ -335,13 +373,26 @@ fn handle_sessions_list(_params: Map) -> ControllerFuture { _ => {} } let pinned = matches!(session.session_id.as_str(), "master" | "subconscious"); - // Roster task line: latest message preview for real instance - // windows; the pinned windows don't need one. + // Roster task line: prefer the v2 `status.detail` (the harness's own + // current-activity line / active tool) when present; otherwise fall + // back to the latest message preview. Pinned windows carry neither. let current_task = if pinned { None } else { - store::latest_message_preview(conn, &session.agent_id, &session.session_id)? - .map(|body| task_preview(&body)) + match session + .current_detail + .as_deref() + .map(str::trim) + .filter(|d| !d.is_empty()) + { + Some(detail) => Some(task_preview(detail)), + None => store::latest_message_preview( + conn, + &session.agent_id, + &session.session_id, + )? + .map(|body| task_preview(&body)), + } }; out.push(summarize(session, unread, pinned, current_task)); } @@ -384,6 +435,7 @@ fn handle_sessions_create(params: Map) -> ControllerFuture { last_seq: 0, created_at: now.clone(), last_message_at: now.clone(), + ..Default::default() }; store::with_connection(&config.workspace_dir, |conn| { store::upsert_session(conn, &session) @@ -400,7 +452,7 @@ fn pinned_placeholder(session_id: &str) -> SessionSummary { agent_id: session_id.to_string(), source: "orchestration".to_string(), harness_type: None, - status: derive_status(true).to_string(), + status: derive_status(None, true).to_string(), current_task: None, label: None, workspace: None, @@ -492,6 +544,7 @@ fn handle_send_master_message(params: Map) -> ControllerFuture { last_seq: seq, created_at: now.clone(), last_message_at: now.clone(), + ..Default::default() }, )?; store::insert_message( @@ -505,6 +558,7 @@ fn handle_send_master_message(params: Map) -> ControllerFuture { body: body.clone(), timestamp: now.clone(), seq, + ..Default::default() }, ) }); @@ -591,6 +645,7 @@ fn handle_send_master_message(params: Map) -> ControllerFuture { last_seq: 0, created_at: now.clone(), last_message_at: now.clone(), + ..Default::default() }, )?; store::insert_message( @@ -604,6 +659,7 @@ fn handle_send_master_message(params: Map) -> ControllerFuture { body: body.clone(), timestamp: now.clone(), seq: 0, + ..Default::default() }, ) }); @@ -713,7 +769,22 @@ fn handle_attention(_params: Map) -> ControllerFuture { } }; - let queue = attention::assemble_attention(approvals, needs_input, unread); + // 4. Remote agent sessions parked on a tool-approval decision (Phase 1's + // persisted `waiting_approval` run-state). Best-effort like the unread + // read: a transient local-DB hiccup must not sink the other sources. + // These fold into the Approval kind (see `assemble_attention`). + let remote_approvals = match store::with_connection( + &config.workspace_dir, + super::ops::gather_remote_approval_signals, + ) { + Ok(remote_approvals) => remote_approvals, + Err(e) => { + log::warn!(target: LOG, "[orchestration_rpc] attention.remote_approvals_failed: {e}"); + Vec::new() + } + }; + + let queue = attention::assemble_attention(approvals, needs_input, unread, remote_approvals); log::debug!( target: LOG, "[orchestration_rpc] attention.exit total={} approvals={} needs_input={} unread={}", @@ -762,13 +833,23 @@ fn handle_self_identity(_params: Map) -> ControllerFuture { } else { let mut rev_params = Map::new(); rev_params.insert("cryptoId".to_string(), Value::from(agent_id.clone())); - match crate::openhuman::tinyplace::handle_tinyplace_directory_reverse(rev_params).await + // Bounded: a slow/degraded relay must degrade this to "no handle", + // never hang the whole card on "Loading identity…" forever. + match tokio::time::timeout( + std::time::Duration::from_secs(5), + crate::openhuman::tinyplace::handle_tinyplace_directory_reverse(rev_params), + ) + .await { - Ok(reverse) => Some(reverse), - Err(e) => { + Ok(Ok(reverse)) => Some(reverse), + Ok(Err(e)) => { log::debug!(target: LOG, "[orchestration_rpc] self_identity reverse miss: {e}"); None } + Err(_) => { + log::warn!(target: LOG, "[orchestration_rpc] self_identity reverse timed out (relay slow)"); + None + } } }; @@ -779,9 +860,16 @@ fn handle_self_identity(_params: Map) -> ControllerFuture { } else { let mut card_params = Map::new(); card_params.insert("agentId".to_string(), Value::from(agent_id.clone())); - crate::openhuman::tinyplace::handle_tinyplace_directory_get_agent(card_params) - .await - .is_ok() + // Bounded like the reverse lookup above: a stalled relay degrades to + // "card not published" rather than pinning the card on "Loading…". + matches!( + tokio::time::timeout( + std::time::Duration::from_secs(5), + crate::openhuman::tinyplace::handle_tinyplace_directory_get_agent(card_params), + ) + .await, + Ok(Ok(_)) + ) }; let identity = super::ops::build_self_identity( @@ -800,6 +888,46 @@ fn handle_self_identity(_params: Map) -> ControllerFuture { }) } +/// Make this agent fully messageable, then echo back the refreshed identity. +/// +/// The `SelfIdentityCard` shows *why* peers can't DM this agent but had no +/// remediation — the user was left to go register a @handle elsewhere even when +/// the wallet already holds one. Two publishes are needed for a peer to actually +/// deliver a first DM, and this does both: +/// +/// 1. **Directory card + encryption key** (`signal_register_encryption_key`): +/// upserts the directory card (minting a minimal one if none exists) and +/// writes the wallet's encryption key into its metadata, so peers can *resolve* +/// this address. +/// 2. **X3DH prekey bundle** (`signal_provision`): generates a signed pre-key + +/// one-time pre-keys and uploads them to `/keys//bundle`, so peers can +/// *initiate* a Signal session. Without this, resolution succeeds but the very +/// first DM 404s on the bundle fetch — "discoverable" would be a lie. This is +/// the step the old card-only publish missed. +/// +/// Then re-reads `self_identity` so the card flips to "discoverable" in place. No +/// @handle registration and no payment: this only publishes presence for the +/// identity the wallet already has. Registering a brand-new paid handle stays in +/// the tiny.place registry surface. +fn handle_publish_identity(_params: Map) -> ControllerFuture { + Box::pin(async move { + log::debug!(target: LOG, "[orchestration_rpc] publish_identity entry"); + // 1. Directory card + encryption key → resolvable. + crate::openhuman::tinyplace::handle_tinyplace_signal_register_encryption_key(Map::new()) + .await + .map_err(|e| format!("publish_identity card: {e}"))?; + // 2. X3DH prekey bundle → a peer can initiate the first DM. A peer that + // resolves the card but finds no bundle 404s on `/keys//bundle`, + // so this is required for real messageability, not optional polish. + crate::openhuman::tinyplace::handle_tinyplace_signal_provision(Map::new()) + .await + .map_err(|e| format!("publish_identity prekeys: {e}"))?; + // 3. Re-read discoverability from the directory so the renderer gets the + // post-publish truth (card + key now live), not an optimistic guess. + handle_self_identity(Map::new()).await + }) +} + /// Relay endpoint + network label for the renderer's relay badge. Reads only the /// configured base URL (no client build / wallet unlock), so it always resolves. fn handle_relay_info(_params: Map) -> ControllerFuture { @@ -870,13 +998,22 @@ mod tests { #[test] fn schemas_use_orchestration_namespace() { let schemas = all_controller_schemas(); - assert_eq!(schemas.len(), 9); + assert_eq!(schemas.len(), 10); assert!(schemas.iter().all(|s| s.namespace == "orchestration")); assert_eq!(schema_for("orchestration_attention").function, "attention"); assert_eq!( schema_for("orchestration_self_identity").function, "self_identity" ); + assert_eq!( + schema_for("orchestration_publish_identity").function, + "publish_identity" + ); + // The publish-identity remediation must be wired into the live registry, + // not just describable — otherwise the SelfIdentityCard button 404s. + assert!(all_registered_controllers() + .iter() + .any(|c| c.schema.function == "publish_identity")); assert_eq!( schema_for("orchestration_relay_info").function, "relay_info" @@ -919,6 +1056,7 @@ mod tests { last_seq: 0, created_at: now.clone(), last_message_at: now, + ..Default::default() }; let resolved = store::with_connection(&config.workspace_dir, |conn| { store::upsert_session(conn, &session)?; @@ -982,9 +1120,59 @@ mod tests { } #[test] - fn status_is_idle_when_active_else_stopped() { - assert_eq!(derive_status(true), "idle"); - assert_eq!(derive_status(false), "stopped"); + fn status_falls_back_to_recency_without_persisted_state() { + // v1/legacy sessions carry no run-state → the original recency heuristic. + assert_eq!(derive_status(None, true), "idle"); + assert_eq!(derive_status(None, false), "stopped"); + // An unrecognised persisted value also falls back to recency. + assert_eq!(derive_status(Some("weird"), true), "idle"); + } + + #[test] + fn status_maps_persisted_v2_run_state() { + assert_eq!(derive_status(Some("running"), true), "running"); + assert_eq!(derive_status(Some("running_tool"), true), "running"); + assert_eq!( + derive_status(Some("waiting_approval"), true), + "waiting-approval" + ); + assert_eq!(derive_status(Some("idle"), true), "idle"); + assert_eq!(derive_status(Some("stopped"), true), "stopped"); + assert_eq!(derive_status(Some("errored"), true), "errored"); + // waiting/errored are honoured even when the session has gone stale. + assert_eq!(derive_status(Some("errored"), false), "errored"); + assert_eq!( + derive_status(Some("waiting_approval"), false), + "waiting-approval" + ); + } + + #[test] + fn status_staleness_downgrades_silent_running_to_stopped() { + // A `running` session that has gone silent (no recent traffic) must not + // sit on a stuck green dot — it downgrades to stopped. + assert_eq!(derive_status(Some("running"), false), "stopped"); + assert_eq!(derive_status(Some("running_tool"), false), "stopped"); + } + + #[test] + fn summarize_reads_persisted_run_state_and_detail() { + let session = OrchestrationSession { + session_id: "w2".to_string(), + agent_id: "@peer".to_string(), + source: "claude".to_string(), + last_seq: 3, + created_at: "2020-01-01T00:00:00Z".to_string(), + // Recent enough to be "active". + last_message_at: chrono::Utc::now().to_rfc3339(), + status_state: Some("waiting_approval".to_string()), + current_detail: Some("approve rm -rf".to_string()), + ..Default::default() + }; + // current_task is threaded from the handler (status.detail preferred there). + let summary = summarize(session, 0, false, Some("approve rm -rf".to_string())); + assert_eq!(summary.status, "waiting-approval"); + assert_eq!(summary.current_task.as_deref(), Some("approve rm -rf")); } #[test] @@ -1010,6 +1198,7 @@ mod tests { created_at: "2020-01-01T00:00:00Z".to_string(), // Stale timestamp → not active → stopped. last_message_at: "2020-01-01T00:00:00Z".to_string(), + ..Default::default() }; let summary = summarize(session, 2, false, Some("drafting cards".to_string())); assert_eq!(summary.harness_type.as_deref(), Some("claude")); diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index 648f5fd92..741b2d46f 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -15,6 +15,9 @@ use super::types::{OrchestrationMessage, OrchestrationSession}; const SCHEMA_DDL: &str = " PRAGMA foreign_keys = ON; + -- `status_state`/`current_detail`/`active_call_id` carry the v2 harness + -- run-state (`status.state`/`status.detail`/`status.active_call_id`). Nullable + -- and additive: a v1/legacy store gets them via `migrate` (existing rows NULL). CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT NOT NULL, agent_id TEXT NOT NULL, @@ -24,9 +27,15 @@ const SCHEMA_DDL: &str = " last_seq INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, last_message_at TEXT NOT NULL, + status_state TEXT, + current_detail TEXT, + active_call_id TEXT, PRIMARY KEY (agent_id, session_id) ); + -- `event_kind`/`tool_name`/`call_id` carry the v2 per-message event shape + -- (`event.kind` + tool identity/correlation). Nullable and additive; v1 and + -- pinned master/subconscious rows leave them NULL. CREATE TABLE IF NOT EXISTS messages ( id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, @@ -35,7 +44,10 @@ const SCHEMA_DDL: &str = " role TEXT NOT NULL, body TEXT NOT NULL, timestamp TEXT NOT NULL, - seq INTEGER NOT NULL DEFAULT 0 + seq INTEGER NOT NULL DEFAULT 0, + event_kind TEXT, + tool_name TEXT, + call_id TEXT ); CREATE INDEX IF NOT EXISTS idx_messages_session @@ -145,6 +157,33 @@ pub fn in_immediate_txn( } } +/// True if `column` already exists on `table` (via `PRAGMA table_info`). Used to +/// make additive `ALTER TABLE ... ADD COLUMN` migrations idempotent — SQLite has +/// no `ADD COLUMN IF NOT EXISTS`, and re-adding an existing column errors. +fn column_exists(conn: &Connection, table: &str, column: &str) -> Result { + // `table` is a hardcoded internal literal (never user input); PRAGMA cannot be + // parameterised, so it is interpolated directly. + let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?; + let mut rows = stmt.query([])?; + while let Some(row) = rows.next()? { + let name: String = row.get(1)?; + if name == column { + return Ok(true); + } + } + Ok(false) +} + +/// Additively add `column` to `table` when it is not already present. Idempotent, +/// so it is safe on a fresh DB (SCHEMA_DDL already created the column → no-op) and +/// on an older store (adds it; existing rows default NULL). +fn add_column_if_missing(conn: &Connection, table: &str, column: &str, decl: &str) -> Result<()> { + if !column_exists(conn, table, column)? { + conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {column} {decl}"))?; + } + Ok(()) +} + /// One-time, `user_version`-gated migrations. Runs after the idempotent /// `SCHEMA_DDL`; each version block executes exactly once per DB. fn migrate(conn: &Connection) -> Result<()> { @@ -177,6 +216,18 @@ fn migrate(conn: &Connection) -> Result<()> { PRAGMA user_version = 1;", )?; } + + // v2 — additive harness-session-v2 receiver columns. Guarded per-column by a + // `table_info` existence check rather than `user_version`, so it is order- and + // freshness-independent: a fresh DB already has them from SCHEMA_DDL (no-op), + // while a v1 store gains them here with existing rows defaulting NULL — which + // `derive_status` reads as "no persisted run-state" and falls back to recency. + add_column_if_missing(conn, "sessions", "status_state", "TEXT")?; + add_column_if_missing(conn, "sessions", "current_detail", "TEXT")?; + add_column_if_missing(conn, "sessions", "active_call_id", "TEXT")?; + add_column_if_missing(conn, "messages", "event_kind", "TEXT")?; + add_column_if_missing(conn, "messages", "tool_name", "TEXT")?; + add_column_if_missing(conn, "messages", "call_id", "TEXT")?; Ok(()) } @@ -196,13 +247,19 @@ pub fn message_exists(conn: &Connection, id: &str) -> Result { pub fn upsert_session(conn: &Connection, s: &OrchestrationSession) -> Result<()> { conn.execute( "INSERT INTO sessions - (session_id, agent_id, source, label, workspace, last_seq, created_at, last_message_at) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) + (session_id, agent_id, source, label, workspace, last_seq, created_at, last_message_at, + status_state, current_detail, active_call_id) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) ON CONFLICT(agent_id, session_id) DO UPDATE SET last_seq = MAX(sessions.last_seq, excluded.last_seq), last_message_at = excluded.last_message_at, label = COALESCE(excluded.label, sessions.label), - workspace = COALESCE(excluded.workspace, sessions.workspace)", + workspace = COALESCE(excluded.workspace, sessions.workspace), + -- Run-state fields COALESCE so a content event (which carries none) + -- never wipes the last status; a fresh `status` event overwrites them. + status_state = COALESCE(excluded.status_state, sessions.status_state), + current_detail = COALESCE(excluded.current_detail, sessions.current_detail), + active_call_id = COALESCE(excluded.active_call_id, sessions.active_call_id)", params![ s.session_id, s.agent_id, @@ -212,6 +269,39 @@ pub fn upsert_session(conn: &Connection, s: &OrchestrationSession) -> Result<()> s.last_seq, s.created_at, s.last_message_at, + s.status_state, + s.current_detail, + s.active_call_id, + ], + )?; + Ok(()) +} + +/// Overwrite a session's v2 run-state columns from an authoritative `status` +/// snapshot. `upsert_session` COALESCEs these so a content event (which carries +/// no run-state) never wipes the last status; a `status` event, by contrast, OWNS +/// them and must be able to CLEAR `current_detail`/`active_call_id` on a +/// `running_tool` → `idle` transition. The row already exists (the ingest path +/// runs `upsert_session` first), so this is a plain UPDATE that SETs — not +/// coalesces — all three, letting `None` clear a stale value. +pub fn apply_run_state( + conn: &Connection, + agent_id: &str, + session_id: &str, + status_state: Option<&str>, + current_detail: Option<&str>, + active_call_id: Option<&str>, +) -> Result<()> { + conn.execute( + "UPDATE sessions + SET status_state = ?3, current_detail = ?4, active_call_id = ?5 + WHERE agent_id = ?1 AND session_id = ?2", + params![ + agent_id, + session_id, + status_state, + current_detail, + active_call_id ], )?; Ok(()) @@ -221,8 +311,9 @@ pub fn upsert_session(conn: &Connection, s: &OrchestrationSession) -> Result<()> pub fn insert_message(conn: &Connection, m: &OrchestrationMessage) -> Result { let changed = conn.execute( "INSERT OR IGNORE INTO messages - (id, agent_id, session_id, chat_kind, role, body, timestamp, seq) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + (id, agent_id, session_id, chat_kind, role, body, timestamp, seq, + event_kind, tool_name, call_id) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", params![ m.id, m.agent_id, @@ -232,6 +323,9 @@ pub fn insert_message(conn: &Connection, m: &OrchestrationMessage) -> Result 0) @@ -291,6 +385,8 @@ pub fn latest_message_preview( .query_row( "SELECT body FROM messages WHERE agent_id = ?1 AND session_id = ?2 + AND (event_kind IS NULL + OR event_kind NOT IN ('status', 'lifecycle', 'unknown')) ORDER BY timestamp DESC, seq DESC LIMIT 1", params![agent_id, session_id], |row| row.get(0), @@ -301,22 +397,12 @@ pub fn latest_message_preview( /// List every persisted session row, newest activity first (stage-7 read surface). pub fn list_sessions(conn: &Connection) -> Result> { let mut stmt = conn.prepare( - "SELECT session_id, agent_id, source, label, workspace, last_seq, created_at, last_message_at + "SELECT session_id, agent_id, source, label, workspace, last_seq, created_at, last_message_at, + status_state, current_detail, active_call_id FROM sessions ORDER BY last_message_at DESC", )?; let rows = stmt - .query_map([], |row| { - Ok(OrchestrationSession { - session_id: row.get(0)?, - agent_id: row.get(1)?, - source: row.get(2)?, - label: row.get(3)?, - workspace: row.get(4)?, - last_seq: row.get(5)?, - created_at: row.get(6)?, - last_message_at: row.get(7)?, - }) - })? + .query_map([], map_session_row)? .collect::, _>>()?; Ok(rows) } @@ -333,8 +419,11 @@ pub fn list_messages_by_session( let rows = match before { Some(before) => { let mut stmt = conn.prepare( - "SELECT id, agent_id, session_id, chat_kind, role, body, timestamp, seq + "SELECT id, agent_id, session_id, chat_kind, role, body, timestamp, seq, + event_kind, tool_name, call_id FROM messages WHERE session_id = ?1 AND timestamp < ?2 + AND (event_kind IS NULL + OR event_kind NOT IN ('status', 'lifecycle', 'unknown')) ORDER BY timestamp DESC, seq DESC LIMIT ?3", )?; let rows = stmt @@ -344,8 +433,11 @@ pub fn list_messages_by_session( } None => { let mut stmt = conn.prepare( - "SELECT id, agent_id, session_id, chat_kind, role, body, timestamp, seq + "SELECT id, agent_id, session_id, chat_kind, role, body, timestamp, seq, + event_kind, tool_name, call_id FROM messages WHERE session_id = ?1 + AND (event_kind IS NULL + OR event_kind NOT IN ('status', 'lifecycle', 'unknown')) ORDER BY timestamp DESC, seq DESC LIMIT ?2", )?; let rows = stmt @@ -359,6 +451,7 @@ pub fn list_messages_by_session( /// Row → [`OrchestrationMessage`] mapper (a free fn so it is `Copy` and can be /// reused across the two `query_map` arms without a borrow-lifetime tangle). +/// Column order MUST match the `SELECT` lists in the message readers. fn map_message_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { let chat_kind: String = row.get(3)?; Ok(OrchestrationMessage { @@ -370,6 +463,27 @@ fn map_message_row(row: &rusqlite::Row<'_>) -> rusqlite::Result) -> rusqlite::Result { + Ok(OrchestrationSession { + session_id: row.get(0)?, + agent_id: row.get(1)?, + source: row.get(2)?, + label: row.get(3)?, + workspace: row.get(4)?, + last_seq: row.get(5)?, + created_at: row.get(6)?, + last_message_at: row.get(7)?, + status_state: row.get(8)?, + current_detail: row.get(9)?, + active_call_id: row.get(10)?, }) } @@ -377,7 +491,9 @@ fn map_message_row(row: &rusqlite::Row<'_>) -> rusqlite::Result Result { let cursor = kv_get(conn, &read_cursor_key(session_id))?.unwrap_or_default(); Ok(conn.query_row( - "SELECT COUNT(*) FROM messages WHERE session_id = ?1 AND timestamp > ?2", + "SELECT COUNT(*) FROM messages WHERE session_id = ?1 AND timestamp > ?2 + AND (event_kind IS NULL + OR event_kind NOT IN ('status', 'lifecycle', 'unknown'))", params![session_id, cursor], |row| row.get(0), )?) @@ -481,21 +597,11 @@ pub fn load_session( session_id: &str, ) -> Result> { conn.query_row( - "SELECT session_id, agent_id, source, label, workspace, last_seq, created_at, last_message_at + "SELECT session_id, agent_id, source, label, workspace, last_seq, created_at, last_message_at, + status_state, current_detail, active_call_id FROM sessions WHERE agent_id = ?1 AND session_id = ?2", params![agent_id, session_id], - |row| { - Ok(OrchestrationSession { - session_id: row.get(0)?, - agent_id: row.get(1)?, - source: row.get(2)?, - label: row.get(3)?, - workspace: row.get(4)?, - last_seq: row.get(5)?, - created_at: row.get(6)?, - last_message_at: row.get(7)?, - }) - }, + map_session_row, ) .optional() .map_err(Into::into) @@ -510,24 +616,13 @@ pub fn list_recent_messages( limit: u32, ) -> Result> { let mut stmt = conn.prepare( - "SELECT id, agent_id, session_id, chat_kind, role, body, timestamp, seq + "SELECT id, agent_id, session_id, chat_kind, role, body, timestamp, seq, + event_kind, tool_name, call_id FROM messages WHERE agent_id = ?1 AND session_id = ?2 ORDER BY timestamp DESC, seq DESC LIMIT ?3", )?; let rows = stmt - .query_map(params![agent_id, session_id, limit], |row| { - let chat_kind: String = row.get(3)?; - Ok(OrchestrationMessage { - id: row.get(0)?, - agent_id: row.get(1)?, - session_id: row.get(2)?, - chat_kind: crate::openhuman::orchestration::types::ChatKind::from_str(&chat_kind), - role: row.get(4)?, - body: row.get(5)?, - timestamp: row.get(6)?, - seq: row.get(7)?, - }) - })? + .query_map(params![agent_id, session_id, limit], map_message_row)? .collect::, _>>()?; // Reverse the DESC scan back to chronological order. Ok(rows.into_iter().rev().collect()) @@ -862,6 +957,7 @@ mod tests { body: "hi".into(), timestamp: "2026-07-02T00:00:00Z".into(), seq, + ..Default::default() } } @@ -875,6 +971,7 @@ mod tests { last_seq: seq, created_at: "2026-07-02T00:00:00Z".into(), last_message_at: "2026-07-02T00:00:00Z".into(), + ..Default::default() } } @@ -919,6 +1016,52 @@ mod tests { .unwrap(); } + #[test] + fn status_lifecycle_unknown_rows_are_hidden_from_thread_and_unread() { + let tmp = tempfile::tempdir().unwrap(); + with_connection(tmp.path(), |conn| { + upsert_session(conn, &session("@a", "h1", 1))?; + + // A v1 row (no event_kind) and typed content rows stay visible; + // status/lifecycle/unknown are persisted (for relay dedup) but must + // not surface in the thread or the unread count. + let mut plain = msg("v1", "@a", "h1", 1); + plain.timestamp = "2026-07-02T00:00:01Z".into(); + insert_message(conn, &plain)?; + + let mut call = msg("call", "@a", "h1", 2); + call.event_kind = Some("tool_call".into()); + call.timestamp = "2026-07-02T00:00:02Z".into(); + insert_message(conn, &call)?; + + for (id, kind, seq) in [ + ("st", "status", 3), + ("lc", "lifecycle", 4), + ("uk", "unknown", 5), + ] { + let mut hidden = msg(id, "@a", "h1", seq); + hidden.event_kind = Some(kind.into()); + hidden.timestamp = format!("2026-07-02T00:00:0{seq}Z"); + insert_message(conn, &hidden)?; + } + + let thread = list_messages_by_session(conn, "h1", 50, None)?; + let ids: Vec<&str> = thread.iter().map(|m| m.id.as_str()).collect(); + assert_eq!(ids, vec!["v1", "call"], "only v1 + typed content rows"); + + // Unread (cursor at 0) counts the two visible rows, not the 3 hidden. + assert_eq!(unread_count(conn, "h1")?, 2); + + // Roster preview skips the hidden rows → newest visible is the call. + assert_eq!( + latest_message_preview(conn, "@a", "h1")?.as_deref(), + Some("hi"), + ); + Ok(()) + }) + .unwrap(); + } + #[test] fn world_diff_is_append_only_with_monotonic_seq_and_idempotent_cycles() { let tmp = tempfile::tempdir().unwrap(); @@ -1239,6 +1382,100 @@ mod tests { .unwrap(); } + #[test] + fn migrates_pre_v2_schema_by_adding_status_and_event_columns() { + // A store created before the harness-session-v2 receiver: the sessions and + // messages tables lack the new run-state / event columns. Opening through + // `with_connection` must add them additively (existing rows read NULL) and + // then accept writes that populate them — proving the ALTER path, not just + // the fresh-DDL path, works. + let tmp = tempfile::tempdir().unwrap(); + let db_path = tmp.path().join("orchestration").join("orchestration.db"); + std::fs::create_dir_all(db_path.parent().unwrap()).unwrap(); + { + let conn = Connection::open(&db_path).unwrap(); + conn.execute_batch( + "CREATE TABLE sessions ( + session_id TEXT NOT NULL, agent_id TEXT NOT NULL, source TEXT NOT NULL, + label TEXT, workspace TEXT, last_seq INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, last_message_at TEXT NOT NULL, + PRIMARY KEY (agent_id, session_id)); + CREATE TABLE messages ( + id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, session_id TEXT NOT NULL, + chat_kind TEXT NOT NULL, role TEXT NOT NULL, body TEXT NOT NULL, + timestamp TEXT NOT NULL, seq INTEGER NOT NULL DEFAULT 0);", + ) + .unwrap(); + // A legacy row predating the new columns. + conn.execute( + "INSERT INTO sessions + (session_id, agent_id, source, last_seq, created_at, last_message_at) + VALUES ('h-old', '@a', 'claude', 1, 't', 't')", + [], + ) + .unwrap(); + conn.execute( + "INSERT INTO messages + (id, agent_id, session_id, chat_kind, role, body, timestamp, seq) + VALUES ('m-old', '@a', 'h-old', 'session', 'agent', 'legacy body', 't', 1)", + [], + ) + .unwrap(); + } + + with_connection(tmp.path(), |conn| { + // New columns now exist on both tables. + for (table, column) in [ + ("sessions", "status_state"), + ("sessions", "current_detail"), + ("sessions", "active_call_id"), + ("messages", "event_kind"), + ("messages", "tool_name"), + ("messages", "call_id"), + ] { + assert!( + column_exists(conn, table, column)?, + "{table}.{column} must be added by migration" + ); + } + + // The legacy rows still read; the new fields come back NULL (None). + let old = load_session(conn, "@a", "h-old")?.expect("legacy session survives"); + assert_eq!(old.source, "claude"); + assert_eq!(old.status_state, None); + assert_eq!(old.current_detail, None); + let msgs = list_recent_messages(conn, "@a", "h-old", 10)?; + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0].body, "legacy body"); + assert_eq!(msgs[0].event_kind, None); + + // And the upgraded schema accepts writes that populate the new fields. + upsert_session( + conn, + &OrchestrationSession { + status_state: Some("running".into()), + current_detail: Some("compiling".into()), + active_call_id: Some("call-1".into()), + ..session("@a", "h-old", 1) + }, + )?; + let updated = load_session(conn, "@a", "h-old")?.unwrap(); + assert_eq!(updated.status_state.as_deref(), Some("running")); + assert_eq!(updated.current_detail.as_deref(), Some("compiling")); + assert_eq!(updated.active_call_id.as_deref(), Some("call-1")); + Ok(()) + }) + .unwrap(); + + // Re-opening is idempotent — the ADD COLUMN guard must not error the second + // time (no `duplicate column name`). + with_connection(tmp.path(), |conn| { + assert!(column_exists(conn, "messages", "call_id")?); + Ok(()) + }) + .unwrap(); + } + #[test] fn in_immediate_txn_serialises_concurrent_seq_allocation() { // Two writers on the same (agent, session) — the drain's inbound persist diff --git a/src/openhuman/orchestration/tools.rs b/src/openhuman/orchestration/tools.rs index c85be4a12..bdf154d49 100644 --- a/src/openhuman/orchestration/tools.rs +++ b/src/openhuman/orchestration/tools.rs @@ -663,6 +663,7 @@ impl Tool for SendToAgentTool { last_seq: seq, created_at: now.clone(), last_message_at: now.clone(), + ..Default::default() }, )?; store::insert_message( @@ -676,6 +677,7 @@ impl Tool for SendToAgentTool { body: message.clone(), timestamp: now.clone(), seq, + ..Default::default() }, ) }); @@ -828,6 +830,7 @@ mod tests { body: body.into(), timestamp: ts.into(), seq, + ..Default::default() }, ) .unwrap(); @@ -845,6 +848,7 @@ mod tests { last_seq: 0, created_at: last_at.into(), last_message_at: last_at.into(), + ..Default::default() }, ) .unwrap(); @@ -898,6 +902,7 @@ mod tests { last_seq: 0, created_at: "2026-07-02T00:01:00Z".into(), last_message_at: "2026-07-02T00:01:00Z".into(), + ..Default::default() }; store::with_connection(&config.workspace_dir, |conn| { store::upsert_session(conn, &sess("@alice", "s-alice"))?; diff --git a/src/openhuman/orchestration/types.rs b/src/openhuman/orchestration/types.rs index 11799abcd..bfd49d99d 100644 --- a/src/openhuman/orchestration/types.rs +++ b/src/openhuman/orchestration/types.rs @@ -153,10 +153,235 @@ impl SessionEnvelopeV1 { } } +// ── v2 envelope: `tinyplace.harness.session.v2` ────────────────────────────── +// +// Hand-rolled mirror of the tiny.place v2 harness wire (snake_case, from the +// TypeScript source), added beside the v1 mirror above. v2 replaces v1's single +// `message` object with a typed `event` carrying a `kind` discriminator + a +// per-kind `payload`, plus a richer `status` run-state. Same +// `scope.wrapper_session_id` routing key as v1. See the port in +// tinyhumansai/tiny.place#210; fold in `tinyplace::types::SessionEnvelopeV2` once +// the vendored SDK ships it. + +/// `envelope_version` discriminator for v2 harness envelopes. +pub const SESSION_ENVELOPE_VERSION_V2: &str = "tinyplace.harness.session.v2"; + +/// The typed payload of a v2 `event`, keyed by `event.kind`. Adjacently tagged on +/// the wire's `kind` (discriminator) + `payload` (content) fields, decoded via +/// [`HarnessEvent::decoded`]. `snake_case` matches the wire kind strings. +/// +/// `tool_kind` inside [`ToolCallPayload`] is intentionally a `String` (not an +/// enum) so an unrecognised tool family does not fail the whole event decode. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "kind", content = "payload", rename_all = "snake_case")] +pub enum HarnessEventKind { + UserPrompt(UserPromptPayload), + AgentMessage(TextPayload), + AgentThinking(TextPayload), + ToolCall(ToolCallPayload), + ToolResult(ToolResultPayload), + ApprovalRequest(ApprovalRequestPayload), + Status(StatusPayload), + Lifecycle(LifecyclePayload), + Error(ErrorPayload), + /// `unknown` wire kind (`{ "raw": any }`) OR any forward-incompatible kind we + /// cannot decode — folded here rather than hard-failing the parse. + Unknown(UnknownPayload), +} + +/// `user_prompt` payload. +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct UserPromptPayload { + #[serde(default)] + pub text: String, + /// `"human"` | `"openhuman_inject"`. + #[serde(default)] + pub source: String, +} + +/// Shared payload for the text-only kinds (`agent_message`, `agent_thinking`). +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct TextPayload { + #[serde(default)] + pub text: String, +} + +/// `tool_call` payload. +#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)] +pub struct ToolCallPayload { + #[serde(default)] + pub call_id: String, + #[serde(default)] + pub tool_name: String, + /// `shell|file_read|file_write|edit|search|web|mcp|task|other` — kept as a + /// free `String` for forward-compatibility (see [`HarnessEventKind`]). + #[serde(default)] + pub tool_kind: String, + #[serde(default)] + pub display: String, + #[serde(default)] + pub input: serde_json::Value, +} + +/// `tool_result` payload. +#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)] +pub struct ToolResultPayload { + #[serde(default)] + pub call_id: String, + #[serde(default)] + pub ok: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub exit_code: Option, + #[serde(default)] + pub is_error: bool, + #[serde(default)] + pub output: String, + #[serde(default)] + pub output_bytes: i64, +} + +/// `approval_request` payload. +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct ApprovalRequestPayload { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub call_id: Option, + #[serde(default)] + pub tool_name: String, + #[serde(default)] + pub display: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reason: Option, +} + +/// `status` payload — the harness run-state. +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct StatusPayload { + /// `running|running_tool|waiting_approval|idle|stopped|errored`. + #[serde(default)] + pub state: String, + #[serde(default)] + pub detail: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub active_call_id: Option, +} + +/// `lifecycle` payload. +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct LifecyclePayload { + /// `session_start|session_end|turn_start|turn_end|compact`. + #[serde(default)] + pub phase: String, +} + +/// `error` payload. +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct ErrorPayload { + #[serde(default)] + pub message: String, + #[serde(default)] + pub fatal: bool, +} + +/// `unknown` payload (`{ "raw": any }`). +#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)] +pub struct UnknownPayload { + #[serde(default)] + pub raw: serde_json::Value, +} + +/// A v2 `event`: common envelope fields + the `kind`/`payload` pair. `kind` and +/// `payload` are kept raw here (a discriminator string + arbitrary JSON) and +/// decoded on demand by [`HarnessEvent::decoded`] so an unknown/garbled event +/// never fails the whole envelope parse (it folds to +/// [`HarnessEventKind::Unknown`]). +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct HarnessEvent { + #[serde(default)] + pub id: String, + #[serde(default)] + pub seq: i64, + #[serde(default)] + pub ts: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub turn_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub model: Option, + /// `owner` (iff `kind == user_prompt`) else `agent`. + #[serde(default)] + pub role: String, + #[serde(default)] + pub kind: String, + #[serde(default)] + pub payload: serde_json::Value, +} + +impl HarnessEvent { + /// Decode `(kind, payload)` into a typed [`HarnessEventKind`]. Forward-safe: + /// an unrecognised `kind` or a payload that fails its kind's schema folds to + /// [`HarnessEventKind::Unknown`] carrying the raw payload, so a future/garbled + /// event never hard-fails the parse (which would silently route it to Master). + pub fn decoded(&self) -> HarnessEventKind { + let tagged = serde_json::json!({ "kind": self.kind, "payload": self.payload }); + serde_json::from_value(tagged).unwrap_or_else(|_| { + HarnessEventKind::Unknown(UnknownPayload { + raw: self.payload.clone(), + }) + }) + } +} + +/// Mirror of the TS `SessionEnvelopeV2` interface. Shares the v1 `bucket`/`scope`/ +/// `harness`/`source` blocks; swaps v1's `message` for a typed [`HarnessEvent`]. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct SessionEnvelopeV2 { + #[serde(default)] + pub envelope_version: String, + #[serde(default)] + pub version: u32, + #[serde(default)] + pub bucket: HarnessBucket, + #[serde(default)] + pub scope: HarnessScope, + #[serde(default)] + pub harness: HarnessInfo, + #[serde(default)] + pub event: HarnessEvent, + #[serde(default)] + pub source: HarnessSource, +} + +impl SessionEnvelopeV2 { + fn is_valid_v2(&self) -> bool { + self.envelope_version == SESSION_ENVELOPE_VERSION_V2 + && !self.scope.harness_session_id.is_empty() + } + + /// Parse a decrypted DM body as a v2 session envelope. Returns `None` for any + /// non-v2 payload (a v1 envelope or a plain DM) so the classifier falls + /// through to v1 then Master. Discriminates purely on `envelope_version`, so a + /// v1 body never matches here (and vice-versa). + pub fn parse(body: &str) -> Option { + let envelope: Self = serde_json::from_str(body).ok()?; + envelope.is_valid_v2().then_some(envelope) + } + + /// The per-pair routing key — identical semantics to v1: the shared + /// `scope.wrapper_session_id`, falling back to `harness_session_id` for a + /// legacy envelope with no per-pair id. + pub fn session_key(&self) -> String { + if !self.scope.wrapper_session_id.is_empty() { + return self.scope.wrapper_session_id.clone(); + } + self.scope.harness_session_id.clone() + } +} + /// Which pinned/session window a persisted message belongs to. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum ChatKind { + /// Safe default (matches [`ChatKind::from_str`]'s unknown fallback). + #[default] Master, Subconscious, Session, @@ -184,7 +409,7 @@ impl ChatKind { /// Durable per-session record. `session_id` is the harness session id for /// [`ChatKind::Session`], or the literal `"master"` for a peer's Master window. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct OrchestrationSession { pub session_id: String, @@ -197,11 +422,22 @@ pub struct OrchestrationSession { pub last_seq: i64, pub created_at: String, pub last_message_at: String, + /// v2 harness run-state from `status.state` + /// (`running|running_tool|waiting_approval|idle|stopped|errored`). `None` for + /// v1/legacy sessions — [`crate`]'s `derive_status` then falls back to recency. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub status_state: Option, + /// v2 `status.detail` — the current-activity line surfaced on the roster. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub current_detail: Option, + /// v2 `status.active_call_id` — the in-flight tool call while running a tool. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub active_call_id: Option, } /// A single persisted message. `body` is DECRYPTED plaintext and therefore /// workspace-internal only. -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct OrchestrationMessage { pub id: String, @@ -212,6 +448,16 @@ pub struct OrchestrationMessage { pub body: String, pub timestamp: String, pub seq: i64, + /// v2 `event.kind` (`user_prompt`/`agent_message`/`tool_call`/…). `None` for + /// v1 and pinned master/subconscious rows, which carry no per-event kind. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub event_kind: Option, + /// v2 tool identifier for `tool_call` / `approval_request` rows. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_name: Option, + /// v2 correlation id linking a `tool_result` back to its `tool_call`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub call_id: Option, } #[cfg(test)] @@ -261,6 +507,208 @@ mod tests { .is_none()); } + // ── v2 envelope ───────────────────────────────────────────────────────── + + /// Build a v2 envelope wire string with the given `kind` + `payload` JSON. + fn v2_wire(kind: &str, payload: &str) -> String { + format!( + r#"{{ + "envelope_version": "tinyplace.harness.session.v2", + "version": 2, + "bucket": {{ "unit": "minute", "start": "s", "end": "e" }}, + "scope": {{ "type": "folder", "key": "repo", "cwd": "/w", + "wrapper_session_id": "w2", "harness_session_id": "h2" }}, + "harness": {{ "provider": "claude", "command": "claude", "argv": [] }}, + "event": {{ "id": "e1", "seq": 4, "ts": "2026-07-05T00:00:00Z", + "turn_id": "t1", "model": "opus", "role": "agent", + "kind": "{kind}", "payload": {payload} }}, + "source": {{ "path": "p", "record_type": "assistant" }} + }}"# + ) + } + + #[test] + fn parses_valid_v2_envelope_and_common_event_fields() { + let wire = v2_wire("agent_message", r#"{ "text": "hi there" }"#); + let env = SessionEnvelopeV2::parse(&wire).expect("valid v2"); + assert_eq!(env.envelope_version, SESSION_ENVELOPE_VERSION_V2); + assert_eq!(env.scope.wrapper_session_id, "w2"); + assert_eq!(env.harness.provider, "claude"); + assert_eq!(env.event.seq, 4); + assert_eq!(env.event.turn_id.as_deref(), Some("t1")); + assert_eq!(env.event.model.as_deref(), Some("opus")); + assert_eq!(env.event.role, "agent"); + assert_eq!(env.session_key(), "w2"); + } + + #[test] + fn v2_decodes_every_event_kind() { + use HarnessEventKind::*; + + let up = SessionEnvelopeV2::parse(&v2_wire( + "user_prompt", + r#"{ "text": "do it", "source": "human" }"#, + )) + .unwrap(); + assert_eq!( + up.event.decoded(), + UserPrompt(UserPromptPayload { + text: "do it".into(), + source: "human".into() + }) + ); + + let am = + SessionEnvelopeV2::parse(&v2_wire("agent_message", r#"{ "text": "ok" }"#)).unwrap(); + assert_eq!( + am.event.decoded(), + AgentMessage(TextPayload { text: "ok".into() }) + ); + + let th = + SessionEnvelopeV2::parse(&v2_wire("agent_thinking", r#"{ "text": "hmm" }"#)).unwrap(); + assert_eq!( + th.event.decoded(), + AgentThinking(TextPayload { text: "hmm".into() }) + ); + + let tc = SessionEnvelopeV2::parse(&v2_wire( + "tool_call", + r#"{ "call_id": "c1", "tool_name": "bash", "tool_kind": "shell", + "display": "ls -la", "input": { "cmd": "ls" } }"#, + )) + .unwrap(); + match tc.event.decoded() { + ToolCall(p) => { + assert_eq!(p.call_id, "c1"); + assert_eq!(p.tool_name, "bash"); + assert_eq!(p.tool_kind, "shell"); + assert_eq!(p.display, "ls -la"); + assert_eq!(p.input["cmd"], "ls"); + } + other => panic!("expected tool_call, got {other:?}"), + } + + let tr = SessionEnvelopeV2::parse(&v2_wire( + "tool_result", + r#"{ "call_id": "c1", "ok": true, "exit_code": 0, "is_error": false, + "output": "done", "output_bytes": 4 }"#, + )) + .unwrap(); + assert_eq!( + tr.event.decoded(), + ToolResult(ToolResultPayload { + call_id: "c1".into(), + ok: true, + exit_code: Some(0), + is_error: false, + output: "done".into(), + output_bytes: 4, + }) + ); + + let ar = SessionEnvelopeV2::parse(&v2_wire( + "approval_request", + r#"{ "call_id": "c9", "tool_name": "rm", "display": "rm -rf x", "reason": "destructive" }"#, + )) + .unwrap(); + assert_eq!( + ar.event.decoded(), + ApprovalRequest(ApprovalRequestPayload { + call_id: Some("c9".into()), + tool_name: "rm".into(), + display: "rm -rf x".into(), + reason: Some("destructive".into()), + }) + ); + + let st = SessionEnvelopeV2::parse(&v2_wire( + "status", + r#"{ "state": "running_tool", "detail": "compiling", "active_call_id": "c1" }"#, + )) + .unwrap(); + assert_eq!( + st.event.decoded(), + Status(StatusPayload { + state: "running_tool".into(), + detail: "compiling".into(), + active_call_id: Some("c1".into()), + }) + ); + + let lc = SessionEnvelopeV2::parse(&v2_wire("lifecycle", r#"{ "phase": "session_end" }"#)) + .unwrap(); + assert_eq!( + lc.event.decoded(), + Lifecycle(LifecyclePayload { + phase: "session_end".into() + }) + ); + + let er = + SessionEnvelopeV2::parse(&v2_wire("error", r#"{ "message": "boom", "fatal": true }"#)) + .unwrap(); + assert_eq!( + er.event.decoded(), + Error(ErrorPayload { + message: "boom".into(), + fatal: true + }) + ); + + let uk = SessionEnvelopeV2::parse(&v2_wire("unknown", r#"{ "raw": { "x": 1 } }"#)).unwrap(); + match uk.event.decoded() { + Unknown(p) => assert_eq!(p.raw["x"], 1), + other => panic!("expected unknown, got {other:?}"), + } + } + + #[test] + fn v2_unrecognised_kind_folds_to_unknown_not_a_parse_error() { + // A future kind the receiver doesn't model must not fail the envelope + // parse (which would silently route the DM to Master); it folds to Unknown + // carrying the raw payload. + let env = SessionEnvelopeV2::parse(&v2_wire("quantum_teleport", r#"{ "flux": 42 }"#)) + .expect("still a valid v2 envelope"); + match env.event.decoded() { + HarnessEventKind::Unknown(p) => assert_eq!(p.raw["flux"], 42), + other => panic!("expected unknown fold, got {other:?}"), + } + } + + #[test] + fn v2_rejects_v1_body_and_plain_and_bad_version() { + // A v1 envelope must NOT parse as v2 (discriminated on envelope_version). + assert!(SessionEnvelopeV2::parse(SAMPLE).is_none()); + assert!(SessionEnvelopeV2::parse("a plain message").is_none()); + // Right shape, wrong version string. + assert!(SessionEnvelopeV2::parse( + r#"{"envelope_version":"tinyplace.harness.session.v3","scope":{"harness_session_id":"h"}}"# + ) + .is_none()); + // Correct version but empty harness id → invalid. + assert!(SessionEnvelopeV2::parse( + r#"{"envelope_version":"tinyplace.harness.session.v2","scope":{"harness_session_id":""}}"# + ) + .is_none()); + // Conversely a v2 body is not a v1 envelope. + let v2 = v2_wire("agent_message", r#"{ "text": "x" }"#); + assert!(SessionEnvelopeV1::parse(&v2).is_none()); + } + + #[test] + fn v2_session_key_falls_back_to_harness_id() { + let env = SessionEnvelopeV2::parse( + r#"{ + "envelope_version": "tinyplace.harness.session.v2", + "scope": { "harness_session_id": "h-only" }, + "event": { "kind": "agent_message", "payload": { "text": "x" } } + }"#, + ) + .expect("valid v2"); + assert_eq!(env.session_key(), "h-only"); + } + #[test] fn session_key_is_the_shared_wrapper_id_then_harness_fallback() { // The single per-pair id lives in `wrapper_session_id`. diff --git a/src/openhuman/tinyplace/manifest.rs b/src/openhuman/tinyplace/manifest.rs index 6022c173c..1f52d8faf 100644 --- a/src/openhuman/tinyplace/manifest.rs +++ b/src/openhuman/tinyplace/manifest.rs @@ -2985,8 +2985,16 @@ pub(crate) fn handle_tinyplace_signal_key_status(_params: Map) -> // AND does it match the current identity key? A stale key (from a // previous wallet) should show as NOT published so the user // re-registers. - let encryption_key_published = match client.directory.get_agent(&agent_id).await { - Ok(card) => { + // Bounded: once a card exists, fetching it hits the relay; a degraded + // relay must degrade this to "not published" (like the Err arm), never + // hang the caller (self_identity → the orchestration identity card). + let card_fetch = tokio::time::timeout( + std::time::Duration::from_secs(5), + client.directory.get_agent(&agent_id), + ) + .await; + let encryption_key_published = match card_fetch { + Ok(Ok(card)) => { let published_key = card .metadata .as_ref() @@ -3008,10 +3016,16 @@ pub(crate) fn handle_tinyplace_signal_key_status(_params: Map) -> log::debug!("{LOG_PREFIX} signal_key_status encryption_key_published={matches}"); matches } - Err(e) => { + Ok(Err(e)) => { log::warn!("{LOG_PREFIX} signal_key_status directory card fetch failed: {e}"); false } + Err(_) => { + log::warn!( + "{LOG_PREFIX} signal_key_status directory card fetch timed out (relay slow)" + ); + false + } }; to_value(serde_json::json!({ diff --git a/src/openhuman/tinyplace/mod.rs b/src/openhuman/tinyplace/mod.rs index b5d9c5ed7..c99f90f04 100644 --- a/src/openhuman/tinyplace/mod.rs +++ b/src/openhuman/tinyplace/mod.rs @@ -32,6 +32,7 @@ pub(crate) use manifest::{ acknowledge_message, decrypt_envelope, ensure_signal_keys_published, handle_tinyplace_contacts_list, handle_tinyplace_directory_get_agent, handle_tinyplace_directory_reverse, handle_tinyplace_signal_key_status, + handle_tinyplace_signal_provision, handle_tinyplace_signal_register_encryption_key, handle_tinyplace_signal_send_message, }; pub(crate) mod ops; diff --git a/vendor/tinyplace b/vendor/tinyplace index 99c6f9c8e..63ce8663b 160000 --- a/vendor/tinyplace +++ b/vendor/tinyplace @@ -1 +1 @@ -Subproject commit 99c6f9c8e282a55134e46ba97148ee9bc5e9e29c +Subproject commit 63ce8663b8ba194d7404b57973116746d0a4eabb