feat(orchestration): receive typed v2 harness-session stream (#4652)

This commit is contained in:
oxoxDev
2026-07-07 23:39:27 -07:00
committed by GitHub
parent 63163ea1be
commit 2dccaf663f
35 changed files with 2271 additions and 156 deletions
@@ -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(<AttentionQueueItem item={remoteApproval} onAction={onAction} />);
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(<AttentionQueueItem item={needsInput} onAction={onAction} />);
@@ -74,4 +74,43 @@ describe('MessageBubble', () => {
render(<MessageBubble message={message({ encrypted: true, body: '••••' })} />);
expect(screen.getByText('••••')).toHaveClass('text-content-muted');
});
it('renders a tool_call as a monospace command row with ▶ + tool name', () => {
const { container } = render(
<MessageBubble message={message({ eventKind: 'tool_call', toolName: 'Bash', body: 'ls' })} />
);
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(
<MessageBubble
message={message({ eventKind: 'tool_result', toolName: 'Bash', body: 'ok' })}
/>
);
expect(screen.getByText('↳')).toBeInTheDocument();
});
it('renders agent_thinking italic + muted with the ∴ glyph', () => {
const { container } = render(
<MessageBubble message={message({ eventKind: 'agent_thinking', body: 'considering…' })} />
);
expect(container.querySelector('p.italic')).not.toBeNull();
expect(screen.getByText('∴')).toBeInTheDocument();
});
it('renders an error row with the ✕ glyph', () => {
render(<MessageBubble message={message({ eventKind: 'error', body: 'boom' })} />);
expect(screen.getByText('✕')).toBeInTheDocument();
});
it('falls back to the plain-dot style for a legacy v1 row (no eventKind)', () => {
const { container } = render(<MessageBubble message={message({ body: 'plain' })} />);
expect(container.querySelector('[data-event-kind="v1"]')).not.toBeNull();
expect(container.querySelector('div.rounded-full')).not.toBeNull();
expect(container.querySelector('p.font-mono')).toBeNull();
});
});
@@ -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 (
<div className="flex gap-2">
<div className="mt-1.5 h-2 w-2 flex-none rounded-full bg-ocean-500" />
<div className="min-w-0 rounded-lg border border-line bg-surface px-3 py-2 shadow-soft">
<div className="flex gap-2" data-event-kind={message.eventKind ?? 'v1'}>
{style.glyph ? (
<span className={`mt-0.5 flex-none text-xs font-semibold ${style.dot}`}>{style.glyph}</span>
) : (
<div
className={`mt-1.5 h-2 w-2 flex-none rounded-full ${style.dot.replace('text-', 'bg-')}`}
/>
)}
<div
className={`min-w-0 rounded-lg border border-line bg-surface px-3 py-2 shadow-soft ${style.accent}`}>
<div className="flex flex-wrap items-center gap-x-2 gap-y-1">
<span className="text-xs font-semibold text-content-secondary">{message.from}</span>
{message.toolName ? (
<span className="rounded bg-surface-strong px-1.5 py-0.5 font-mono text-[10px] text-content-secondary">
{message.toolName}
</span>
) : null}
<span className="text-[10px] text-content-faint">{formatTime(message.timestamp)}</span>
</div>
<p
className={`mt-1 whitespace-pre-wrap break-words text-sm ${
message.encrypted ? 'text-content-muted' : 'text-content'
}`}>
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}
</p>
</div>
@@ -37,6 +37,9 @@ const props = (over: Partial<OrchestrationSidebarProps>): OrchestrationSidebarPr
steeringText: null,
selfIdentity: null,
identityLoading: false,
onPublishIdentity: vi.fn(),
publishingIdentity: false,
publishIdentityError: null,
attentionQueue: null,
attentionLoading: false,
onAttentionAction: vi.fn(),
@@ -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}
</div>
<SelfIdentityCard identity={selfIdentity} loading={identityLoading} />
<SelfIdentityCard
identity={selfIdentity}
loading={identityLoading}
onPublish={onPublishIdentity}
publishing={publishingIdentity}
publishError={publishIdentityError}
/>
<AttentionQueueView
queue={attentionQueue}
@@ -60,6 +60,66 @@ describe('SelfIdentityCard', () => {
).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(<SelfIdentityCard identity={undiscoverable} loading={false} onPublish={onPublish} />);
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(
<SelfIdentityCard identity={undiscoverable} loading={false} onPublish={vi.fn()} publishing />
);
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(
<SelfIdentityCard
identity={undiscoverable}
loading={false}
onPublish={vi.fn()}
publishError="boom"
/>
);
expect(screen.getByTestId('tinyplace-self-identity-publish-error')).toHaveTextContent(
'tinyplaceOrchestration.identity.publishFailed'
);
});
it('shows no "make discoverable" button once discoverable', () => {
render(<SelfIdentityCard identity={discoverable} loading={false} onPublish={vi.fn()} />);
expect(screen.queryByTestId('tinyplace-self-identity-publish')).toBeNull();
});
it('offers a "Republish keys" action while discoverable that fires onPublish', () => {
const onPublish = vi.fn();
render(<SelfIdentityCard identity={discoverable} loading={false} onPublish={onPublish} />);
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(<SelfIdentityCard identity={undiscoverable} loading={false} />);
expect(screen.queryByTestId('tinyplace-self-identity-publish')).toBeNull();
});
it('copies the address to the clipboard on click', async () => {
render(<SelfIdentityCard identity={discoverable} loading={false} />);
fireEvent.click(screen.getByTestId('tinyplace-self-identity-copy'));
@@ -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({
</span>
</div>
{identity.discoverable && onPublish ? (
<div className="mt-2 flex flex-wrap items-center gap-2">
<button
type="button"
data-testid="tinyplace-self-identity-republish"
onClick={onPublish}
disabled={publishing}
className="inline-flex items-center rounded-md border border-line px-2 py-0.5 text-[10px] font-medium text-content-muted transition hover:bg-surface-hover disabled:opacity-50">
{publishing
? t('tinyplaceOrchestration.identity.publishing')
: t('tinyplaceOrchestration.identity.republish')}
</button>
{publishError ? (
<span
data-testid="tinyplace-self-identity-republish-error"
className="text-[10px] text-coral-600 dark:text-coral-300">
{t('tinyplaceOrchestration.identity.publishFailed')}
</span>
) : null}
</div>
) : null}
{!identity.discoverable ? (
<p className="mt-2 rounded-md bg-coral-50 px-2 py-1 text-[10px] text-coral-700 dark:bg-coral-500/10 dark:text-coral-300">
{t('tinyplaceOrchestration.identity.undiscoverableHint')}
</p>
<div className="mt-2 rounded-md bg-coral-50 px-2 py-1.5 dark:bg-coral-500/10">
<p className="text-[10px] text-coral-700 dark:text-coral-300">
{t('tinyplaceOrchestration.identity.undiscoverableHint')}
</p>
{onPublish ? (
<button
type="button"
data-testid="tinyplace-self-identity-publish"
onClick={onPublish}
disabled={publishing}
className="mt-1.5 inline-flex items-center rounded-md bg-coral-600 px-2 py-1 text-[11px] font-semibold text-white transition hover:bg-coral-700 disabled:opacity-50 dark:bg-coral-500 dark:hover:bg-coral-400">
{publishing
? t('tinyplaceOrchestration.identity.publishing')
: t('tinyplaceOrchestration.identity.makeDiscoverable')}
</button>
) : null}
{publishError ? (
<p
data-testid="tinyplace-self-identity-publish-error"
className="mt-1 text-[10px] text-coral-700 dark:text-coral-300">
{t('tinyplaceOrchestration.identity.publishFailed')}
</p>
) : null}
</div>
) : null}
</section>
);
@@ -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(),
},
@@ -69,6 +69,10 @@ export default function TinyPlaceOrchestrationTab() {
// the whole tab.
const [selfIdentity, setSelfIdentity] = useState<SelfIdentity | null>(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<string | null>(null);
const [relayInfo, setRelayInfo] = useState<RelayInfo | null>(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}
+5 -1
View File
@@ -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': 'منشور',
+5 -1
View File
@@ -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': 'প্রকাশিত',
+6 -1
View File
@@ -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',
+5 -1
View File
@@ -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',
+5 -1
View File
@@ -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',
+5 -1
View File
@@ -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 dannuaire 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 dannuaire',
'tinyplaceOrchestration.identity.key': 'Clé de chiffrement',
'tinyplaceOrchestration.identity.published': 'Publié',
+5 -1
View File
@@ -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': 'प्रकाशित',
+5 -1
View File
@@ -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',
+5 -1
View File
@@ -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',
+5 -1
View File
@@ -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': '게시됨',
+5 -1
View File
@@ -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',
+5 -1
View File
@@ -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',
+5 -1
View File
@@ -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': 'Опубликовано',
+5 -1
View File
@@ -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': '已发布',
@@ -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<SelfIdentity>('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<SelfIdentity>('openhuman.orchestration_publish_identity', {}),
/** The relay endpoint + network label the core is talking to (RelayBadge). */
relayInfo: () => call<RelayInfo>('openhuman.orchestration_relay_info', {}),
};
@@ -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 } : {}),
};
}
+220 -2
View File
@@ -141,6 +141,25 @@ pub(crate) struct UnreadSignal {
pub last_message_at: Option<String>,
}
/// 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<String>,
/// The approval prompt / current-activity line (PII-safe summary text).
pub detail: Option<String>,
/// 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<String>,
pub created_at: Option<String>,
}
// ── 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:<request_id>`) 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<ApprovalSignal>,
needs_input: Vec<NeedsInputSignal>,
unread: Vec<UnreadSignal>,
remote_approvals: Vec<RemoteApprovalSignal>,
) -> AttentionQueue {
let mut items: Vec<AttentionItem> = 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<super::types::OrchestrationSession>,
) -> Vec<RemoteApprovalSignal> {
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<AttentionKind> = 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"]
);
}
}
+607 -45
View File
@@ -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<String>,
workspace: Option<String>,
/// 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<String>,
/// v2 tool identity for `tool_call` / `approval_request`.
tool_name: Option<String>,
/// v2 `call_id` correlating `tool_result` → `tool_call`.
call_id: Option<String>,
/// 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<String>,
/// v2 `status.detail` → the session's `current_detail` (roster task line).
status_detail: Option<String>,
/// v2 `status.active_call_id` → the session's `active_call_id`.
active_call_id: Option<String>,
/// 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<String>,
call_id: Option<String>,
status_state: Option<String>,
status_detail: Option<String>,
active_call_id: Option<String>,
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<String> {
(!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<i64> = 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();
+35
View File
@@ -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<Vec<super::attention::RemoteApprovalSignal>> {
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))?;
+218 -29
View File
@@ -33,6 +33,7 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
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<RegisteredController> {
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<String> {
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<String, Value>) -> 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<String, Value>) -> 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<String, Value>) -> 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<String, Value>) -> ControllerFuture {
body: body.clone(),
timestamp: now.clone(),
seq,
..Default::default()
},
)
});
@@ -591,6 +645,7 @@ fn handle_send_master_message(params: Map<String, Value>) -> 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<String, Value>) -> ControllerFuture {
body: body.clone(),
timestamp: now.clone(),
seq: 0,
..Default::default()
},
)
});
@@ -713,7 +769,22 @@ fn handle_attention(_params: Map<String, Value>) -> 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<String, Value>) -> 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<String, Value>) -> 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<String, Value>) -> 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/<addr>/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<String, Value>) -> 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/<addr>/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<String, Value>) -> 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"));
+286 -49
View File
@@ -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<T>(
}
}
/// 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<bool> {
// `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<bool> {
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<bool> {
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<boo
m.body,
m.timestamp,
m.seq,
m.event_kind,
m.tool_name,
m.call_id,
],
)?;
Ok(changed > 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<Vec<OrchestrationSession>> {
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::<std::result::Result<Vec<_>, _>>()?;
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<OrchestrationMessage> {
let chat_kind: String = row.get(3)?;
Ok(OrchestrationMessage {
@@ -370,6 +463,27 @@ fn map_message_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<OrchestrationMes
body: row.get(5)?,
timestamp: row.get(6)?,
seq: row.get(7)?,
event_kind: row.get(8)?,
tool_name: row.get(9)?,
call_id: row.get(10)?,
})
}
/// Row → [`OrchestrationSession`] mapper. Column order MUST match the `SELECT`
/// lists in [`list_sessions`] and [`load_session`].
fn map_session_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<OrchestrationSession> {
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<OrchestrationMes
pub fn unread_count(conn: &Connection, session_id: &str) -> Result<i64> {
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<Option<OrchestrationSession>> {
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<Vec<OrchestrationMessage>> {
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::<std::result::Result<Vec<_>, _>>()?;
// 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
+5
View File
@@ -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"))?;
+451 -3
View File
@@ -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<i64>,
#[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<String>,
#[serde(default)]
pub tool_name: String,
#[serde(default)]
pub display: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reason: Option<String>,
}
/// `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<String>,
}
/// `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<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub model: Option<String>,
/// `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<Self> {
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<String>,
/// v2 `status.detail` — the current-activity line surfaced on the roster.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub current_detail: Option<String>,
/// 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<String>,
}
/// 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<String>,
/// v2 tool identifier for `tool_call` / `approval_request` rows.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tool_name: Option<String>,
/// v2 correlation id linking a `tool_result` back to its `tool_call`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub call_id: Option<String>,
}
#[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`.
+17 -3
View File
@@ -2985,8 +2985,16 @@ pub(crate) fn handle_tinyplace_signal_key_status(_params: Map<String, Value>) ->
// 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<String, Value>) ->
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!({
+1
View File
@@ -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;