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