Skip to content

fix: bind queued messages to session owners #452

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chubes4 merged 4 commits into from
Jul 21, 2026
Merged

fix: bind queued messages to session owners #452

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4b5932d
fix: bind queued messages to session owners
chubes4 Jul 21, 2026
239dec7
fix: bind chat sessions to canonical owners
chubes4 Jul 21, 2026
9ccb897
fix: atomically bind chat session owners
chubes4 Jul 21, 2026
5943fd7
fix: authorize chat runs from session store
chubes4 Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions src/Channels/register-agents-chat-ability.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,10 +178,14 @@ function agents_chat_dispatch( array $input ) {
$agent = agents_chat_optional_string( $input['agent'] ?? null ) ?? '';
if ( null !== $session_id ) {
try {
WP_Agent_Chat_Run_Control::start_run( $run_id, $session_id, array( 'agent' => $agent ), $run_context['workspace'], $run_context['owner'] );
$started = WP_Agent_Chat_Run_Control::start_run( $run_id, $session_id, array( 'agent' => $agent ), $run_context['workspace'], $run_context['owner'], $run_context['conversation_store'] );
} catch ( \RuntimeException $error ) {
return new \WP_Error( 'agents_chat_run_workspace_unsupported', $error->getMessage() );
}
if ( is_wp_error( $started ) ) {
do_action( 'agents_chat_dispatch_failed', $started->get_error_code(), $input );
return $started;
}
}

$result = call_user_func( $handler, $input );
Expand Down Expand Up @@ -220,7 +224,11 @@ function agents_chat_dispatch( array $input ) {
$resolved_session_id = agents_chat_optional_string( $result['session_id'] ?? null ) ?? $session_id;
if ( null !== $resolved_session_id ) {
if ( null === $session_id ) {
WP_Agent_Chat_Run_Control::start_run( $result_run_id, $resolved_session_id, array( 'agent' => $agent ), $run_context['workspace'], $run_context['owner'] );
$started = WP_Agent_Chat_Run_Control::start_run( $result_run_id, $resolved_session_id, array( 'agent' => $agent ), $run_context['workspace'], $run_context['owner'], $run_context['conversation_store'] );
if ( is_wp_error( $started ) ) {
do_action( 'agents_chat_dispatch_failed', $started->get_error_code(), $input );
return $started;
}
}

$status = WP_Agent_Run_Outcome::run_control_status( $result );
Expand Down
23 changes: 22 additions & 1 deletion src/Channels/register-agents-chat-run-control-abilities.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,13 +232,23 @@ function agents_queue_chat_message( array $input ) {
if ( $context['workspace'] instanceof WP_Agent_Workspace_Scope ) {
$input['workspace'] = $context['workspace']->to_array();
}
try {
if ( ! WP_Agent_Chat_Run_Control::can_queue_message( $input, $context['workspace'], $context['owner'], $context['conversation_store'] ) ) {
return agents_chat_run_control_no_handler( 'agents_chat_run_not_found', 'No chat run was found for the requested session owner.' );
}
} catch ( \RuntimeException $error ) {
return new \WP_Error( 'agents_chat_run_workspace_unsupported', $error->getMessage() );
}
if ( is_array( $context['owner'] ) ) {
$input['session_owner'] = $context['owner'];
}

$handler = apply_filters( 'wp_agent_chat_message_queue_handler', null, $input );
if ( is_callable( $handler ) ) {
$result = agents_chat_run_control_normalize_result( call_user_func( $handler, $input ), 'agents_chat_message_queue_invalid_result' );
} else {
try {
$result = WP_Agent_Chat_Run_Control::queue_message( $input, $context['workspace'], $context['owner'] );
$result = WP_Agent_Chat_Run_Control::queue_message( $input, $context['workspace'], $context['owner'], $context['conversation_store'] );
} catch ( \InvalidArgumentException|\RuntimeException $error ) {
return new \WP_Error( 'agents_chat_message_queue_invalid_result', $error->getMessage() );
}
Expand Down Expand Up @@ -280,6 +290,17 @@ function agents_chat_run_read_permission( array $input ): bool {
/** @param array<string, mixed> $input Ability input. */
function agents_chat_run_enqueue_permission( array $input ): bool {
$allowed = agents_chat_run_write_permission( $input );
$context = WP_Agent_Chat_Run_Control::context_from_input( $input );
if ( $allowed && ! is_wp_error( $context ) ) {
try {
$allowed = WP_Agent_Chat_Run_Control::can_queue_message( $input, $context['workspace'], $context['owner'], $context['conversation_store'] );
} catch ( \RuntimeException $error ) {
unset( $error );
$allowed = false;
}
} else {
$allowed = false;
}
$allowed = (bool) apply_filters( 'agents_chat_run_enqueue_permission', $allowed, $input );
return (bool) apply_filters( 'agents_chat_run_control_permission', $allowed, $input );
}
Expand Down
1 change: 1 addition & 0 deletions src/Channels/register-default-agents-chat-handler.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,6 +184,7 @@ public static function execute( array $input ) {
'agent_slug' => $agent_slug,
'run_id' => $runtime_context['run_id'],
'principal' => $input['principal'] ?? null,
'conversation_store' => $store,
'tool_executor_registry' => $executor_registry,
),
);
Expand Down
187 changes: 169 additions & 18 deletions src/Runtime/class-wp-agent-chat-run-control.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@

namespace AgentsAPI\AI;

use AgentsAPI\Core\Database\Chat\WP_Agent_Conversation_Sessions;
use AgentsAPI\Core\Database\Chat\WP_Agent_Conversation_Store;
use AgentsAPI\Core\Workspace\WP_Agent_Workspace_Scope;

defined( 'ABSPATH' ) || exit;
Expand Down Expand Up @@ -47,10 +49,10 @@ public static function statuses(): array {
}

/**
* Resolve and bind canonical workspace and principal ownership.
* Resolve canonical workspace, principal, and conversation store context.
*
* @param array<mixed> $input Ability input.
* @return array{workspace:?WP_Agent_Workspace_Scope,owner:?array{type:string,key:string}}|\WP_Error
* @return array{workspace:?WP_Agent_Workspace_Scope,owner:?array{type:string,key:string},conversation_store:?WP_Agent_Conversation_Store}|\WP_Error
*/
public static function context_from_input( array $input ) {
$workspace = null;
Expand Down Expand Up @@ -112,8 +114,9 @@ public static function context_from_input( array $input ) {
}

return array(
'workspace' => $workspace,
'owner' => $owner,
'workspace' => $workspace,
'owner' => $owner,
'conversation_store' => WP_Agent_Conversation_Sessions::get_store( WP_Agent_Run_Control::string_keyed_array( $input ) ),
);
}

Expand Down Expand Up @@ -182,16 +185,27 @@ public static function normalize_status( mixed $status ): string {
* @param string $session_id Session ID.
* @param array<string,mixed> $metadata Run metadata.
* @param array<string,mixed>|null $owner Canonical conversation owner.
* @return array<string,mixed> Normalized run.
* @return array<string,mixed>|\WP_Error Normalized run.
*/
public static function start_run( string $run_id, string $session_id, array $metadata = array(), ?WP_Agent_Workspace_Scope $workspace = null, ?array $owner = null ): array {
public static function start_run( string $run_id, string $session_id, array $metadata = array(), ?WP_Agent_Workspace_Scope $workspace = null, ?array $owner = null, ?WP_Agent_Conversation_Store $conversation_store = null ) {
$canonical_owner = self::session_owner_fingerprint( $session_id, $workspace, $owner, $conversation_store );
if ( $canonical_owner instanceof \WP_Error ) {
return $canonical_owner;
}

$state = self::state( $workspace );
$existing = $state['runs'][ $run_id ] ?? null;
if ( is_array( $existing ) && ( $session_id !== self::string_value( $existing['session_id'] ?? '' ) || ! self::fingerprint_matches( $existing['_owner'] ?? null, $canonical_owner ) ) ) {
return new \WP_Error( 'agents_chat_run_owner_forbidden', 'The run_id is already bound to another session or owner.' );
}

return self::normalize_run( WP_Agent_Run_Control::start_run(
self::OPTION_KEY,
$run_id,
array(
'session_id' => $session_id,
'metadata' => $metadata,
'_owner' => self::owner_fingerprint( $owner ),
'_owner' => $canonical_owner,
),
$workspace
) );
Expand Down Expand Up @@ -265,15 +279,20 @@ public static function cancellation_interrupt_for_run( string $run_id, string $s
*
* @param array<string,mixed> $input Canonical queue input.
* @param array<string,mixed>|null $owner Canonical conversation owner.
* @return array<string,mixed> Queue result.
* @return array<string,mixed>|\WP_Error Queue result.
*/
public static function queue_message( array $input, ?WP_Agent_Workspace_Scope $workspace = null, ?array $owner = null ): array {
public static function queue_message( array $input, ?WP_Agent_Workspace_Scope $workspace = null, ?array $owner = null, ?WP_Agent_Conversation_Store $conversation_store = null ) {
$session_id = trim( self::string_value( $input['session_id'] ?? null ) );
$run_id = trim( self::string_value( $input['run_id'] ?? null ) );
if ( '' === $session_id ) {
throw new \InvalidArgumentException( 'session_id must be a non-empty string' );
}

$target = self::queue_target( $input, $workspace, $owner, $conversation_store );
if ( $target instanceof \WP_Error ) {
return $target;
}
$run_id = $target['run_id'];

$queued_id = 'queued_' . str_replace( 'run_', '', self::generate_run_id() );
$item = array(
'queued_message_id' => $queued_id,
Expand All @@ -284,7 +303,7 @@ public static function queue_message( array $input, ?WP_Agent_Workspace_Scope $w
'attachments' => is_array( $input['attachments'] ?? null ) ? $input['attachments'] : array(),
'client_context' => is_array( $input['client_context'] ?? null ) ? $input['client_context'] : array(),
'created_at' => self::now(),
'_owner' => self::owner_fingerprint( $owner ),
'_owner' => $target['owner_fingerprint'],
);

$state = self::state( $workspace );
Expand All @@ -294,7 +313,7 @@ public static function queue_message( array $input, ?WP_Agent_Workspace_Scope $w
self::save_state( $state, $workspace );

return self::normalize_run( array(
'run_id' => '' !== $run_id ? $run_id : self::generate_run_id(),
'run_id' => $run_id,
'session_id' => $session_id,
'status' => self::STATUS_QUEUED,
'updated_at' => self::now(),
Expand All @@ -310,20 +329,32 @@ public static function queue_message( array $input, ?WP_Agent_Workspace_Scope $w
* @param array<string,mixed>|null $owner Canonical conversation owner.
* @return array<int,array<string,mixed>> Queued items.
*/
public static function claim_queued_messages( string $session_id, ?WP_Agent_Workspace_Scope $workspace = null, ?array $owner = null ): array {
public static function claim_queued_messages( string $session_id, ?WP_Agent_Workspace_Scope $workspace = null, ?array $owner = null, ?WP_Agent_Conversation_Store $conversation_store = null ): array {
$canonical_owner = self::session_owner_fingerprint( $session_id, $workspace, $owner, $conversation_store );
if ( $canonical_owner instanceof \WP_Error ) {
return array();
}

$state = self::state( $workspace );
$items = array_values( array_filter(
$state['queues'][ $session_id ] ?? array(),
static fn( array $item ): bool => self::owner_matches( $item['_owner'] ?? '', $owner )
static fn( array $item ): bool => self::fingerprint_matches( $item['_owner'] ?? null, $canonical_owner )
) );
if ( count( $items ) !== count( $state['queues'][ $session_id ] ?? array() ) ) {
return array();
}
unset( $state['queues'][ $session_id ] );
self::save_state( $state, $workspace );
return array_map( array( self::class, 'public_queue_item' ), $items );
}

/**
* Check whether queue input resolves to a session owned by the principal.
*
* @param array<string,mixed> $input Canonical queue input.
* @param array<string,mixed>|null $owner Canonical conversation owner.
*/
public static function can_queue_message( array $input, ?WP_Agent_Workspace_Scope $workspace = null, ?array $owner = null, ?WP_Agent_Conversation_Store $conversation_store = null ): bool {
return ! ( self::queue_target( $input, $workspace, $owner, $conversation_store ) instanceof \WP_Error );
}

/**
* List lifecycle events for a principal-owned run.
*
Expand Down Expand Up @@ -399,7 +430,127 @@ private static function owner_fingerprint( ?array $owner ): string {
/** @param array<string,mixed>|null $owner */
private static function owner_matches( mixed $stored, ?array $owner ): bool {
$stored = is_string( $stored ) ? $stored : '';
return '' === $stored || ( '' !== self::owner_fingerprint( $owner ) && hash_equals( $stored, self::owner_fingerprint( $owner ) ) );
$fingerprint = self::owner_fingerprint( $owner );
return '' === $stored ? '' === $fingerprint : '' !== $fingerprint && hash_equals( $stored, $fingerprint );
}

private static function fingerprint_matches( mixed $stored, string $expected ): bool {
return is_string( $stored ) && hash_equals( $expected, $stored );
}

/**
* Resolve a queue target from stored run state, which is authoritative for
* both the session identity and its canonical owner.
*
* @param array<string,mixed> $input Queue input.
* @param array<string,mixed>|null $owner Resolved principal owner.
* @return array{run_id:string,owner_fingerprint:string}|\WP_Error
*/
private static function queue_target( array $input, ?WP_Agent_Workspace_Scope $workspace, ?array $owner, ?WP_Agent_Conversation_Store $conversation_store ) {
$session_id = trim( self::string_value( $input['session_id'] ?? null ) );
$run_id = trim( self::string_value( $input['run_id'] ?? null ) );
$state = self::state( $workspace );
$canonical_owner = self::session_owner_fingerprint( $session_id, $workspace, $owner, $conversation_store );
if ( $canonical_owner instanceof \WP_Error ) {
return $canonical_owner;
}

if ( '' !== $run_id ) {
$candidate = $state['runs'][ $run_id ] ?? null;
if ( ! is_array( $candidate ) || $session_id !== trim( self::string_value( $candidate['session_id'] ?? null ) ) || ! self::fingerprint_matches( $candidate['_owner'] ?? null, $canonical_owner ) ) {
return new \WP_Error( 'agents_chat_run_not_found', 'No chat run was found for the requested session owner.' );
}
} else {
$latest_key = '';
foreach ( $state['runs'] as $candidate_run_id => $candidate ) {
if ( $session_id === trim( self::string_value( $candidate['session_id'] ?? null ) ) && self::fingerprint_matches( $candidate['_owner'] ?? null, $canonical_owner ) ) {
$candidate_key = self::string_value( $candidate['updated_at'] ?? '' ) . ':' . self::string_value( $candidate['started_at'] ?? '' ) . ':' . $candidate_run_id;
if ( '' !== $run_id && $candidate_key <= $latest_key ) {
continue;
}
$latest_key = $candidate_key;
$run_id = $candidate_run_id;
}
}
}

if ( '' === $run_id ) {
return new \WP_Error( 'agents_chat_run_not_found', 'No chat run was found for the requested session.' );
}

return array(
'run_id' => $run_id,
'owner_fingerprint' => $canonical_owner,
);
}

/**
* Resolve ownership from the canonical conversation store.
*
* @param array<string,mixed>|null $owner Resolved principal owner.
* @return string|\WP_Error Canonical owner fingerprint.
*/
private static function session_owner_fingerprint( string $session_id, ?WP_Agent_Workspace_Scope $workspace, ?array $owner, ?WP_Agent_Conversation_Store $conversation_store ) {
$session_id = trim( $session_id );
$fingerprint = self::owner_fingerprint( $owner );
$conversation_store ??= WP_Agent_Conversation_Sessions::get_store();
if ( '' === $session_id ) {
return new \WP_Error( 'agents_chat_run_not_found', 'No chat session was requested.' );
}
if ( $workspace instanceof WP_Agent_Workspace_Scope && '' === $fingerprint ) {
return new \WP_Error( 'agents_chat_run_owner_required', 'Explicit workspace run control requires an authenticated conversation owner.' );
}

$canonical_workspace = $workspace;
if ( null === $canonical_workspace && $conversation_store instanceof WP_Agent_Conversation_Store && is_array( $owner ) ) {
$canonical_workspace = WP_Agent_Workspace_Scope::from_parts( 'site', function_exists( 'get_current_blog_id' ) ? (string) get_current_blog_id() : 'default' );
}

if ( $canonical_workspace instanceof WP_Agent_Workspace_Scope && $conversation_store instanceof WP_Agent_Conversation_Store ) {
if ( ! is_array( $owner ) ) {
return new \WP_Error( 'agents_chat_run_owner_forbidden', 'The canonical conversation session is not owned by this workspace principal.' );
}
$canonical_owner = array(
'type' => self::string_value( $owner['type'] ?? '' ),
'key' => self::string_value( $owner['key'] ?? '' ),
);
if ( ! is_array( WP_Agent_Conversation_Sessions::get_owned_session( $conversation_store, $session_id, $canonical_workspace, $canonical_owner ) ) ) {
return new \WP_Error( 'agents_chat_run_owner_forbidden', 'The canonical conversation session is not owned by this workspace principal.' );
}
return $fingerprint;
}

if ( $workspace instanceof WP_Agent_Workspace_Scope ) {
if ( ! $conversation_store instanceof WP_Agent_Conversation_Store ) {
return new \WP_Error( 'agents_chat_run_session_store_required', 'Explicit workspace run control requires an authoritative conversation session store.' );
}
}

// Omitted-workspace compatibility is limited to ownership already proven
// by a site-local run. New principal-owned sessions need a canonical store.
$historical_owners = array();
foreach ( self::state( $workspace )['runs'] as $run ) {
if ( $session_id !== self::string_value( $run['session_id'] ?? '' ) ) {
continue;
}
$stored = is_string( $run['_owner'] ?? null ) ? $run['_owner'] : '';
$historical_owners[ $stored ] = true;
}

if ( count( $historical_owners ) > 1 ) {
return new \WP_Error( 'agents_chat_run_owner_forbidden', 'The session has conflicting historical owner state.' );
}
if ( 1 === count( $historical_owners ) ) {
$stored = (string) array_key_first( $historical_owners );
if ( ! self::fingerprint_matches( $stored, $fingerprint ) ) {
return new \WP_Error( 'agents_chat_run_owner_forbidden', 'The session is owned by another conversation principal.' );
}
$fingerprint = $stored;
} elseif ( '' !== $fingerprint ) {
return new \WP_Error( 'agents_chat_run_session_store_required', 'Principal-owned run control requires an authoritative conversation session store.' );
}

return $fingerprint;
}

/** @param array<string,mixed>|null $owner */
Expand Down
Loading