fix(ci): stop FINAL reviews deadlocking on outsider-flagged PRs

On an outsider-flagged PR Claude answered the FINAL request with a
security-only comment (no Phase 1 table). The gate only counts a reply with
the Phase 1 / 'Final architecture review' heading, and a FINAL requested after
HEAD is not 'stale', so recovery never re-asked: 9 PRs sat on 'FINAL Claude not
complete on HEAD' with green CI.

- merge-gate.js: a FINAL that Claude answered without the review is stale, so
  the hourly recovery re-requests it (max 3 per HEAD).
- claude.yml: the outsider preamble tells Claude to keep the requested FINAL
  format and add security findings as an extra section.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

.github/scripts/merge-gate-selftest.js

@@ -21,6 +21,24 @@ const finals = [
 ];
 assert('stale when head after final', mg.isStaleFinalAfterPush(finals, '2026-06-12T09:00:00Z'));
 
+// PR #6056 shape: FINAL requested after HEAD; Claude answered with a security-only
+// comment (no Phase 1 table). That FINAL never completes, so recovery must re-ask.
+{
+  const head = '2026-10-27T20:40:21Z';
+  const trigger = { user: { login: 'MervinPraison' }, body: mg.FINAL_CLAUDE_REVIEW_BODY, created_at: '2026-10-27T20:42:06Z' };
+  const securityOnly = { user: { login: 'praisonai-triage-agent[bot]' }, body: '## Security and architecture review at `fec9380dd`. Origin: outsider\n\n**Verdict:** PASS', created_at: '2026-10-27T20:50:17Z' };
+  const realFinal = { user: { login: 'praisonai-triage-agent[bot]' }, body: '#### Phase 1 — AGENTS.md review\n| Check | Result |', created_at: '2026-10-27T20:55:00Z' };
+  const gateSecurity = { user: { login: 'praisonai-triage-agent[bot]' }, body: 'SECURITY_REVIEW: PASS\n| file | category |', created_at: '2026-10-27T20:50:17Z' };
+  assert('FINAL answered without the review is stale (re-ask)', mg.isStaleFinalAfterPush([trigger, securityOnly], head));
+  assert('FINAL answered without the review is not complete', !mg.finalClaudeCompletedOnSha([trigger, securityOnly], head));
+  assert('a real FINAL reply after a security-only one completes it', !mg.isStaleFinalAfterPush([trigger, securityOnly, realFinal], head)
+    && mg.finalClaudeCompletedOnSha([trigger, securityOnly, realFinal], head));
+  assert('the gate security review comment is not a FINAL answer', !mg.isStaleFinalAfterPush([trigger, gateSecurity], head));
+  assert('no reply yet: FINAL in progress, not stale', !mg.isStaleFinalAfterPush([trigger], head));
+  const retries = [0, 1, 2].map((i) => ({ ...trigger, created_at: `2026-10-27T2${i + 1}:00:00Z` }));
+  assert('re-asking stops after the per-HEAD cap', !mg.isStaleFinalAfterPush([...retries, { ...securityOnly, created_at: '2026-10-27T23:30:00Z' }], head));
+}
+
 const withRecovery = [
   ...finals,
   { user: { login: 'github-actions[bot]' }, body: '@claude FINAL architecture reviewer', created_at: '2026-06-12T09:30:00Z' },

.github/scripts/merge-gate.js

@@ -322,6 +322,7 @@ function isStaleFinalAfterPush(comments, headPushedAt) {
     new Date(a.created_at) > new Date(b.created_at) ? a : b
   );
   const finalTime = new Date(latestFinal.created_at).getTime();
+  if (finalAnsweredWithoutReview(comments, finalTime, headTime)) return true;
   if (headTime <= finalTime + 60000) return false;
 
   // The finished reply must follow BOTH the latest FINAL trigger and HEAD, so
@@ -342,6 +343,31 @@ function isStaleFinalAfterPush(comments, headPushedAt) {
   return !claudeSinceHead;
 }
 
+// FINAL re-requests allowed for one HEAD before a maintainer has to look.
+const FINAL_RETRIES_PER_HEAD = 3;
+
+/**
+ * Claude answered the latest FINAL, but not with the review the gate needs (no
+ * Phase 1 table / "Final architecture review" heading), e.g. a security-only
+ * comment on an outsider-origin PR. That FINAL never completes, and because it
+ * was requested after HEAD it is not "stale" either, so without this the PR
+ * waits forever. Treat it as stale so recovery asks again (capped per HEAD).
+ */
+function finalAnsweredWithoutReview(comments, finalTime, headTime) {
+  const after = comments.filter((c) => new Date(c.created_at).getTime() > finalTime);
+  if (after.some(isClaudeFinalReplyComment)) return false;
+  const retries = comments.filter(
+    (c) => isFinalClaudeTriggerComment(c) && new Date(c.created_at).getTime() >= headTime - 60000
+  ).length;
+  if (retries >= FINAL_RETRIES_PER_HEAD) return false;
+  return after.some((c) => {
+    if (!CLAUDE_AUTOMATION_LOGINS.has(c.user?.login || '') || AUTO_ACTORS.includes(c.user?.login)) return false;
+    const body = c.body || '';
+    if (isClaudeTriggerNoise(c) || /^\s*@claude\b/i.test(body) || /^\s*SECURITY_REVIEW:/m.test(body)) return false;
+    return true;
+  });
+}
+
 function needsStaleFinalRecovery(comments, headPushedAt) {
   return (
     hasFinalClaudeReviewTrigger(comments) &&

.github/workflows/claude.yml

@@ -380,7 +380,7 @@ jobs:
             if (process.env.IS_FORK === 'true') {
               preamble = `${untrusted}\n\nThis is a fork PR. Do a security review as well as your normal review:\n${checklist}\nStart your single comment with a line "Security review: PASS" or "Security review: FAIL" plus the findings (advisory: a maintainer merges fork PRs), then your normal review.`;
             } else if (isPr) {
-              preamble = `${untrusted}\n\nThis PR was built from or commented on by outsiders. ${rules} Also check the diff against this list and report concerns:\n${checklist}\nThe merge gate runs an independent security review before merging.`;
+              preamble = `${untrusted}\n\nThis PR was built from or commented on by outsiders. ${rules} Also check the diff against this list and report concerns:\n${checklist}\nThe merge gate runs an independent security review before merging. If the request asks for a review in a set format (the FINAL architecture review's Phase 1 table and Phase 3 verdict), keep that format exactly and add these security findings as an extra section; the merge gate only recognises the requested format.`;
             } else {
               preamble = `${untrusted}\n\nYou may implement this outsider's request, under these rules. ${rules} An independent security review runs before any merge.`;
             }

src/praisonai-agents/praisonaiagents/agent/handoff.py

@@ -190,6 +190,39 @@ def _get_handoff_depth() -> int:
     """Get current handoff depth."""
     return len(_get_handoff_chain())
 
+def _tool_pair_safe_tail(messages: List[Any], n: int) -> List[Any]:
+    """Return the last ``n`` messages without splitting tool_calls/tool pairs.
+
+    A count-based slice can cut between an assistant ``tool_calls`` message and
+    its ``tool`` responses; providers reject the resulting orphans. Drop tool
+    results whose call was cut off, and assistant tool_calls whose responses
+    are incomplete (together with their partial responses).
+    """
+    if n <= 0:
+        return []
+    tail = list(messages[-n:])
+
+    def _call_ids(msg: Any) -> set:
+        calls = msg.get('tool_calls') if isinstance(msg, dict) else None
+        return {c.get('id') if isinstance(c, dict) else getattr(c, 'id', None) for c in calls or ()}
+
+    def _is_tool(msg: Any) -> bool:
+        return isinstance(msg, dict) and msg.get('role') == 'tool'
+
+    response_ids = {m.get('tool_call_id') for m in tail if _is_tool(m)}
+    dropped: set = set()
+    kept = []
+    for msg in tail:
+        ids = _call_ids(msg)
+        if ids and not ids <= response_ids:
+            dropped |= ids
+            continue
+        kept.append(msg)
+    call_ids = set().union(*(_call_ids(m) for m in kept))
+    return [m for m in kept
+            if not (_is_tool(m) and (m.get('tool_call_id') in dropped
+                                     or m.get('tool_call_id') not in call_ids))]
+
 def _push_handoff(agent_name: str) -> None:
     """Push agent to handoff chain.
 
@@ -702,22 +735,22 @@ class Handoff:
         # Apply context policy
         if self.config.context_policy == ContextPolicy.NONE:
             filtered_messages = []
-        elif self.config.context_policy == ContextPolicy.LAST_N:
-            n = self.config.max_context_messages
+        elif self.config.context_policy in (ContextPolicy.LAST_N, ContextPolicy.SUMMARY):
             if self.config.preserve_system:
                 system_msgs = [m for m in messages if isinstance(m, dict) and m.get('role') == 'system']
-                other_msgs = [m for m in messages if not (isinstance(m, dict) and m.get('role') == 'system')]
-                filtered_messages = system_msgs + other_msgs[-n:]
+                base = [m for m in messages if not (isinstance(m, dict) and m.get('role') == 'system')]
             else:
-                filtered_messages = messages[-n:]
-        elif self.config.context_policy == ContextPolicy.SUMMARY:
-            # For summary, keep system + last few messages
-            if self.config.preserve_system:
-                system_msgs = [m for m in messages if isinstance(m, dict) and m.get('role') == 'system']
-                other_msgs = [m for m in messages if not (isinstance(m, dict) and m.get('role') == 'system')]
-                filtered_messages = system_msgs + other_msgs[-3:]  # Last 3 non-system
+                system_msgs, base = [], list(messages)
+            if self.config.context_policy == ContextPolicy.LAST_N:
+                tail = _tool_pair_safe_tail(base, self.config.max_context_messages)
             else:
-                filtered_messages = messages[-3:]
+                # SUMMARY: the first user message (the task goal) + last few
+                # messages. An LLM summary stays opt-in via ``input_filter``.
+                tail = _tool_pair_safe_tail(base, 3)
+                goal = next((m for m in base if isinstance(m, dict) and m.get('role') == 'user'), None)
+                if goal is not None and not any(m is goal for m in tail):
+                    tail = [goal] + tail
+            filtered_messages = system_msgs + tail
         else:  # FULL
             filtered_messages = messages[:]
         

src/praisonai-agents/praisonaiagents/agent/image_agent.py

@@ -14,7 +14,7 @@ import warnings
 warnings.filterwarnings("ignore", "Valid config keys have changed in V2", UserWarning)
 class ImageGenerationConfig(BaseModel):
     """Configuration for image generation settings."""
-    style: str = Field(default="natural", description="Style of the generated image")
+    style: Optional[str] = Field(default=None, description="Style of the generated image (only sent when set)")
     response_format: str = Field(default="url", description="Format of the response (url or b64_json)")
     timeout: int = Field(default=600, description="Timeout in seconds for the API call")
     api_base: Optional[str] = Field(default=None, description="Optional API base URL")
@@ -36,7 +36,7 @@ class ImageAgent(Agent):
         backstory: Optional[str] = None,
         instructions: Optional[str] = None,
         llm: Optional[Union[str, Any]] = None,
-        style: str = "natural",
+        style: Optional[str] = None,
         response_format: str = "url",
         timeout: int = 600,
         api_base: Optional[str] = None,

src/praisonai-agents/praisonaiagents/bots/protocols.py

@@ -246,7 +246,11 @@ class ChannelDescriptor(Protocol):
     secrets and plain config keys). A descriptor MAY additionally expose an
     optional ``setup(io)`` hook for bespoke, multi-step flows the field list
     cannot express; the wizard calls it when present and merges the returned
-    values. Because it is optional, a minimal descriptor (like the example
+    values. It MAY likewise expose an optional ``standalone_sender`` — a (sync or async)
+    ``(target: DeliveryTarget, text: str) -> None`` that raises on failure — so
+    scheduled ``deliver:`` targets reach the channel even when no gateway is
+    resident (``praisonai schedule tick`` from OS cron / CI / scale-to-zero).
+    Because these are optional, a minimal descriptor (like the example
     below) omits it — so this Protocol is intentionally *not* used as an
     ``isinstance`` type guard; consumers read attributes with ``getattr``
     fallbacks instead.

src/praisonai-agents/praisonaiagents/scheduler/config_store.py

@@ -15,7 +15,7 @@ import time
 from typing import Dict, Iterator, List, Optional
 
 from .models import ScheduleJob, RunRecord
-from .due import is_due as _is_due, is_misfire as _is_misfire, resolve_schedule_timezone
+from .due import is_due as _is_due, is_misfire as _is_misfire, lease_state as _lease_state, past_until as _past_until, resolve_schedule_timezone
 from .hook_emit import emit_schedule_add, emit_schedule_remove
 
 logger = get_logger(__name__)
@@ -187,25 +187,37 @@ class ConfigYamlScheduleStore:
         job that is due and not already leased by someone else we:
 
         * advance ``last_run_at`` to ``now`` (pre-advancing the schedule so a
-          later poll no longer sees it as due — at-least-once → at-most-once),
+          later poll no longer sees it as due),
         * take a lease (``lease_until = now + lease_seconds``, ``lease_owner``),
           persisting both in the same atomic write,
-        * remove one-shot jobs (``delete_after_run``) immediately so no other
-          ticker can pick them up.
+        * mark spent jobs (one-shot, ``max_runs`` met) as retiring instead of
+          deleting them; the lease keeps competitors out and ``mark_run``
+          removes them once the run completes.
 
-        A crashed run leaves a lease that expires after ``lease_seconds``; the
-        job then becomes due again and is retried. Losers of the race simply do
-        not see the job in their returned list.
+        A lease that expires without ``complete`` is an abandoned (crashed)
+        run and is recorded as ``abandoned`` history. A retiring job is then
+        re-offered once (at-least-once across crashes) and retired if that
+        attempt is abandoned too; a recurring job is at-most-once per slot and
+        next runs at its following slot. Losers of the race simply do not see
+        the job in their returned list.
         """
         claimed: List[ScheduleJob] = []
-        # One-shots auto-deleted by this claim. Emitted after the lock below;
-        # a later ``remove(job.id)`` by the runner is then a no-op, so a spent
-        # one-shot yields exactly one SCHEDULE_REMOVE.
+        # Spent jobs retired by this claim. Emitted after the lock below.
         auto_removed: List[ScheduleJob] = []
-        # (job_id, job_name) for occurrences suppressed by the misfire policy.
-        # Logged as ``missed`` history after the lock so a dropped occurrence is
-        # observable instead of vanishing silently.
+        # (job_id, job_name, status) for occurrences suppressed by the misfire
+        # policy (``missed``) or runs whose lease expired (``abandoned``).
+        # Logged after the lock so a lost run is observable, not silent.
         missed: List[tuple] = []
+
+        def take(job: ScheduleJob) -> None:
+            job._lease_until = now + lease_seconds
+            job._lease_owner = owner_id
+            # ``run_count`` is counted at claim; flag it so ``mark_run`` does
+            # not double-count the same fire.
+            job._run_counted = True
+            self._held_leases[job.id] = owner_id
+            claimed.append(job)
+
         with self._lock, self._file_lock():
             # Re-read from disk so we observe cross-process claims/leases.
             self._reload_locked()
@@ -213,10 +225,31 @@ class ConfigYamlScheduleStore:
             for job in list(self._jobs.values()):
                 if not job.enabled:
                     continue
-                lease_until = getattr(job, "_lease_until", 0.0) or 0.0
-                lease_owner = getattr(job, "_lease_owner", None)
-                # An unexpired lease held by another owner blocks the claim.
-                if lease_until > now and lease_owner != owner_id:
+                state = _lease_state(job, now, owner_id)
+                if state == "busy" or (
+                    # Our own spent-job run outlived its lease but is still in
+                    # flight: re-offering it would fire it twice.
+                    state == "abandoned"
+                    and getattr(job, "_retiring", 0)
+                    and self._held_leases.get(job.id) == owner_id
+                ):
+                    continue
+                if state == "abandoned":
+                    missed.append((job.id, job.name, "abandoned"))
+                    job._lease_until = 0.0
+                    job._lease_owner = None
+                    changed = True
+                    if getattr(job, "_retiring", 0) == 1 and not _past_until(job, now, self._default_timezone):
+                        # First attempt of a spent job crashed: retry once
+                        # (unless its ``until`` deadline has passed meanwhile).
+                        job._retiring = 2
+                        take(job)
+                        continue
+                if getattr(job, "_retiring", 0):
+                    # Run finished (or its retry was abandoned too): retire.
+                    auto_removed.append(self._jobs.pop(job.id))
+                    self._job_state.pop(job.id, None)
+                    changed = True
                     continue
                 if not _is_due(job, now, self._default_timezone):
                     # A job past its ``until`` end instant is not due, but must
@@ -235,26 +268,17 @@ class ConfigYamlScheduleStore:
                 # again on the next poll.
                 if _is_misfire(job, now, self._default_timezone):
                     job.last_run_at = now
-                    missed.append((job.id, job.name))
+                    missed.append((job.id, job.name, "missed"))
                     changed = True
                     continue
                 # Win the claim: pre-advance + lease atomically.
                 job.last_run_at = now
-                # Count the fire here (needed so ``should_retire()`` can drop a
-                # spent job before a competitor reclaims it) and flag it so the
-                # later ``mark_run`` does not double-count the same fire.
                 job.run_count += 1
-                job._run_counted = True
-                job._lease_until = now + lease_seconds
-                job._lease_owner = owner_id
-                self._held_leases[job.id] = owner_id
-                claimed.append(job)
-                changed = True
                 if job.should_retire(now):
-                    # Spent (one-shot, run-count met, or past ``until``): remove
-                    # now so no competitor re-claims it.
-                    auto_removed.append(self._jobs.pop(job.id))
-                    self._job_state.pop(job.id, None)
+                    # Spent: keep it (leased) so a crash cannot lose it.
+                    job._retiring = 1
+                take(job)
+                changed = True
             if changed:
                 if not self._save():
                     # The lease/last_run_at advance was NOT persisted (full
@@ -270,10 +294,10 @@ class ConfigYamlScheduleStore:
             emit_schedule_remove(job)
         # Record suppressed occurrences so a missed run is visible in history
         # rather than silently dropped (log_run takes only the thread lock).
-        for job_id, job_name in missed:
+        for job_id, job_name, status in missed:
             self.log_run(
                 job_id=job_id,
-                status="missed",
+                status=status,
                 job_name=job_name,
             )
         return claimed

src/praisonai-agents/praisonaiagents/scheduler/due.py

@@ -228,6 +228,26 @@ def quota_hold_from_failure(
     return now + float(retry_after) + max(float(slack_seconds), 0.0)
 
 
+def past_until(job: Any, now: float, default_timezone: str | None = None) -> bool:
+    """Return ``True`` when ``now`` is at or past the job's ``until`` end instant."""
+    until = getattr(job, "until", None)
+    if until is None:
+        return False
+    try:
+        until_ts = localize_wall_clock(
+            datetime.fromisoformat(until),
+            job.schedule.tz or default_timezone,
+        ).timestamp()
+    except (ValueError, TypeError):
+        # Fail safe: a stop bound that cannot be parsed must NOT let the
+        # job run past it forever. Treat it as elapsed; the claim path then
+        # retires it via ``ScheduleJob.should_retire`` (which returns True
+        # for an unparseable ``until``) rather than firing it.
+        logger.warning("Invalid 'until' timestamp for job %s: %s", job.id, until)
+        return True
+    return now >= until_ts
+
+
 def is_due(
     job: Any,
     now: float,
@@ -264,22 +284,8 @@ def is_due(
     if hold_until is not None and now < hold_until:
         return False
 
-    until = getattr(job, "until", None)
-    if until is not None:
-        try:
-            until_ts = localize_wall_clock(
-                datetime.fromisoformat(until),
-                sched.tz or default_timezone,
-            ).timestamp()
-        except (ValueError, TypeError):
-            # Fail safe: a stop bound that cannot be parsed must NOT let the
-            # job run past it forever. Treat the job as not-due; the claim path
-            # then retires it via ``ScheduleJob.should_retire`` (which returns
-            # True for an unparseable ``until``) rather than firing it.
-            logger.warning("Invalid 'until' timestamp for job %s: %s", job.id, until)
-            return False
-        if now >= until_ts:
-            return False
+    if past_until(job, now, default_timezone):
+        return False
 
     if sched.kind == "every":
         if sched.every_seconds is None:
@@ -341,3 +347,26 @@ def is_due(
         return now >= next_run
 
     return False
+
+
+def lease_state(job: Any, now: float, owner_id: str) -> str | None:
+    """Classify a job's claim lease for an atomic ``claim_due``.
+
+    Returns:
+        ``"busy"`` when another owner holds an unexpired lease, or this owner
+        still holds one on a spent (retiring) job whose run is in flight;
+        ``"abandoned"`` when a lease expired without ``complete`` — the run
+        crashed or the process died mid-run; ``"done"`` when a spent job's
+        run finished (lease released) but it was not removed yet; ``None``
+        when the job is free to be evaluated for due-ness.
+    """
+    lease_until = getattr(job, "_lease_until", 0.0) or 0.0
+    lease_owner = getattr(job, "_lease_owner", None)
+    retiring = getattr(job, "_retiring", 0)
+    if lease_owner is not None and lease_until > now:
+        if lease_owner != owner_id or retiring:
+            return "busy"
+        return None
+    if lease_owner is not None:
+        return "abandoned"
+    return "done" if retiring else None

src/praisonai-agents/praisonaiagents/scheduler/models.py

@@ -215,7 +215,7 @@ class RunRecord:
         job_id: ID of the job that was executed.
         job_name: Human-readable name of the job.
         status: Execution status. One of ``"succeeded"``, ``"failed"``,
-                ``"skipped"``, ``"no_change"``, or ``"missed"``. ``"no_change"``
+                ``"skipped"``, ``"no_change"``, ``"missed"`` or ``"abandoned"``. ``"no_change"``
                 is the stateful "monitor mode" outcome — a watched source was
                 unchanged since the last tick, so the model turn was suppressed
                 silently (no tokens, no delivery). It is distinct from
@@ -225,6 +225,8 @@ class RunRecord:
                 the process was down and fell outside the job's
                 ``misfire_grace_seconds`` — it was not run, but is recorded
                 (not silently dropped) so an operator can see the gap.
+                ``"abandoned"`` records a claimed run whose lease expired
+                without completing (crash or restart mid-run).
         result: Agent response text (truncated if very long).
         error: Error message if status is ``"failed"``.
         duration: Wall-clock seconds for execution.
@@ -234,7 +236,7 @@ class RunRecord:
 
     job_id: str
     job_name: str = ""
-    status: Literal["succeeded", "failed", "skipped", "no_change", "missed"] = "succeeded"
+    status: Literal["succeeded", "failed", "skipped", "no_change", "missed", "abandoned"] = "succeeded"
     result: Optional[str] = None
     error: Optional[str] = None
     duration: float = 0.0
@@ -460,7 +462,7 @@ class ScheduleJob:
         every claim path so the retire decision is defined once. A job retires
         when it is:
 
-        * a one-shot (``delete_after_run``),
+        * a one-shot (``delete_after_run``) that has fired,
         * at or past its ``max_runs`` budget, or
         * past its ``until`` end instant (when ``now`` is supplied).
 
@@ -478,7 +480,8 @@ class ScheduleJob:
                 ``until`` check is skipped (callers that only ran a job and
                 want the run-count/one-shot decision).
         """
-        if self.delete_after_run:
+        # A one-shot that has not fired yet is waiting, not spent.
+        if self.delete_after_run and self.run_count > 0:
             return True
         if self.max_runs is not None and self.run_count >= self.max_runs:
             return True
@@ -587,6 +590,11 @@ class ScheduleJob:
         if lease_until:
             d["lease_until"] = lease_until
             d["lease_owner"] = getattr(self, "_lease_owner", None)
+        # Spent job claimed for its final run: 1 = first attempt, 2 = the one
+        # retry after a crashed (abandoned) attempt. Omitted otherwise.
+        retiring = getattr(self, "_retiring", 0)
+        if retiring:
+            d["retiring"] = retiring
         return d
 
     @classmethod
@@ -645,4 +653,5 @@ class ScheduleJob:
         # Restore atomic-claim lease metadata if present (see ``to_dict``).
         job._lease_until = d.get("lease_until", 0.0) or 0.0
         job._lease_owner = d.get("lease_owner")
+        job._retiring = d.get("retiring", 0) or 0
         return job

src/praisonai-agents/praisonaiagents/scheduler/runner.py

@@ -62,8 +62,10 @@ class ScheduleRunner:
 
         Args:
             owner_id: Stable identifier for this ticker (process/host).
-            lease_seconds: Lease window; a claim not completed expires after
-                this so a crashed run is eventually retried.
+            lease_seconds: Lease window. A claim not completed within it is
+                recorded as an ``abandoned`` run: a one-shot / spent job is
+                retried once (at-least-once), a recurring job runs at its next
+                slot (at-most-once per slot).
         """
         claim = getattr(self._store, "claim_due", None)
         if callable(claim):
@@ -85,7 +87,11 @@ class ScheduleRunner:
         duration: float = 0.0,
         delivered: bool = False,
     ) -> None:
-        """Update ``last_run_at``, log history, and optionally delete one-shot jobs.
+        """Update ``last_run_at``, log history, and retire spent jobs.
+
+        A spent job (one-shot, ``max_runs`` met) is removed after a run that
+        did not fail; a failed one is kept, disabled, so it stays visible in
+        ``schedule list`` and can be re-enabled or removed by the user.
 
         Args:
             job: The job that was executed.
@@ -97,10 +103,9 @@ class ScheduleRunner:
         """
         job.last_run_at = time.time()
         # Count this fire exactly once. An atomic ``claim_due`` already
-        # incremented ``run_count`` when it reserved the job (and needs to, so
-        # ``should_retire()`` can drop a spent job before a competitor reclaims
-        # it); it flags the job with ``_run_counted`` so we do not double-count
-        # here. The non-atomic ``get_due_jobs`` path never sets that flag, so
+        # incremented ``run_count`` when it reserved the job (so
+        # ``should_retire()`` can mark a spent job as retiring); it flags the
+        # job with ``_run_counted`` so we do not double-count here. The non-atomic ``get_due_jobs`` path never sets that flag, so
         # ``mark_run`` remains the single incrementer there.
         if getattr(job, "_run_counted", False):
             job._run_counted = False
@@ -122,9 +127,19 @@ class ScheduleRunner:
         # The run already happened (its claim was persisted), so a failed write
         # here is reported, not raised: one bad write must not stop the loop.
         try:
-            if job.should_retire(job.last_run_at):
+            if not job.should_retire(job.last_run_at):
+                self._store.update(job)
+            elif status != "failed":
                 self._store.remove(job.id)
             else:
+                # Keep a failed spent job visible (disabled) instead of
+                # deleting it, so the user can see and re-run it. Undo the
+                # spent fire (history keeps the failed row) so resuming it
+                # runs it again instead of retiring it unrun.
+                job.enabled = False
+                job._retiring = 0
+                job.run_count = max(0, job.run_count - 1)
+                job.last_run_at = None
                 self._store.update(job)
         except OSError as e:
             logger.error("Failed to persist run state for job %r: %s", job.id, e)

src/praisonai-agents/praisonaiagents/scheduler/store.py

@@ -14,7 +14,7 @@ import time
 from typing import Dict, Iterator, List, Optional
 
 from .models import ScheduleJob, RunRecord
-from .due import is_due as _is_due, is_misfire as _is_misfire
+from .due import is_due as _is_due, is_misfire as _is_misfire, lease_state as _lease_state, past_until as _past_until
 from .hook_emit import emit_schedule_add, emit_schedule_remove
 from ..utils.atomic_io import atomic_write_json
 
@@ -175,25 +175,37 @@ class FileScheduleStore:
         job that is due and not already leased by someone else we:
 
         * advance ``last_run_at`` to ``now`` (pre-advancing the schedule so a
-          later poll no longer sees it as due — at-least-once → at-most-once),
+          later poll no longer sees it as due),
         * take a lease (``lease_until = now + lease_seconds``, ``lease_owner``),
           persisting both in the same atomic write,
-        * remove one-shot jobs (``delete_after_run``) immediately so no other
-          ticker can pick them up.
+        * mark spent jobs (one-shot, ``max_runs`` met) as retiring instead of
+          deleting them; the lease keeps competitors out and ``mark_run``
+          removes them once the run completes.
 
-        A crashed run leaves a lease that expires after ``lease_seconds``; the
-        job then becomes due again and is retried. Losers of the race simply do
-        not see the job in their returned list.
+        A lease that expires without ``complete`` is an abandoned (crashed)
+        run and is recorded as ``abandoned`` history. A retiring job is then
+        re-offered once (at-least-once across crashes) and retired if that
+        attempt is abandoned too; a recurring job is at-most-once per slot and
+        next runs at its following slot. Losers of the race simply do not see
+        the job in their returned list.
         """
         claimed: List[ScheduleJob] = []
-        # One-shots auto-deleted by this claim. Emitted after the lock below;
-        # a later ``remove(job.id)`` by the runner is then a no-op, so a spent
-        # one-shot yields exactly one SCHEDULE_REMOVE.
+        # Spent jobs retired by this claim. Emitted after the lock below.
         auto_removed: List[ScheduleJob] = []
-        # (job_id, job_name) for occurrences suppressed by the misfire policy.
-        # Logged as ``missed`` history after the lock so a dropped occurrence is
-        # observable instead of vanishing silently.
+        # (job_id, job_name, status) for occurrences suppressed by the misfire
+        # policy (``missed``) or runs whose lease expired (``abandoned``).
+        # Logged after the lock so a lost run is observable, not silent.
         missed: List[tuple] = []
+
+        def take(job: ScheduleJob) -> None:
+            job._lease_until = now + lease_seconds
+            job._lease_owner = owner_id
+            # ``run_count`` is counted at claim; flag it so ``mark_run`` does
+            # not double-count the same fire.
+            job._run_counted = True
+            self._held_leases[job.id] = owner_id
+            claimed.append(job)
+
         with self._lock, self._file_lock():
             # Re-read from disk so we observe cross-process claims/leases.
             self._reload_locked()
@@ -201,10 +213,30 @@ class FileScheduleStore:
             for job in list(self._jobs.values()):
                 if not job.enabled:
                     continue
-                lease_until = getattr(job, "_lease_until", 0.0) or 0.0
-                lease_owner = getattr(job, "_lease_owner", None)
-                # An unexpired lease held by another owner blocks the claim.
-                if lease_until > now and lease_owner != owner_id:
+                state = _lease_state(job, now, owner_id)
+                if state == "busy" or (
+                    # Our own spent-job run outlived its lease but is still in
+                    # flight: re-offering it would fire it twice.
+                    state == "abandoned"
+                    and getattr(job, "_retiring", 0)
+                    and self._held_leases.get(job.id) == owner_id
+                ):
+                    continue
+                if state == "abandoned":
+                    missed.append((job.id, job.name, "abandoned"))
+                    job._lease_until = 0.0
+                    job._lease_owner = None
+                    changed = True
+                    if getattr(job, "_retiring", 0) == 1 and not _past_until(job, now):
+                        # First attempt of a spent job crashed: retry once
+                        # (unless its ``until`` deadline has passed meanwhile).
+                        job._retiring = 2
+                        take(job)
+                        continue
+                if getattr(job, "_retiring", 0):
+                    # Run finished (or its retry was abandoned too): retire.
+                    auto_removed.append(self._jobs.pop(job.id))
+                    changed = True
                     continue
                 if not _is_due(job, now):
                     # A job past its ``until`` end instant is not due, but must
@@ -224,25 +256,17 @@ class FileScheduleStore:
                     # poll does not see it as due again) and record the miss,
                     # without claiming or firing.
                     job.last_run_at = now
-                    missed.append((job.id, job.name))
+                    missed.append((job.id, job.name, "missed"))
                     changed = True
                     continue
                 # Win the claim: pre-advance + lease atomically.
                 job.last_run_at = now
-                # Count the fire here (needed so ``should_retire()`` can drop a
-                # spent job before a competitor reclaims it) and flag it so the
-                # later ``mark_run`` does not double-count the same fire.
                 job.run_count += 1
-                job._run_counted = True
-                job._lease_until = now + lease_seconds
-                job._lease_owner = owner_id
-                self._held_leases[job.id] = owner_id
-                claimed.append(job)
-                changed = True
                 if job.should_retire(now):
-                    # Spent (one-shot, run-count met, or past ``until``): remove
-                    # now so no competitor re-claims it.
-                    auto_removed.append(self._jobs.pop(job.id))
+                    # Spent: keep it (leased) so a crash cannot lose it.
+                    job._retiring = 1
+                take(job)
+                changed = True
             if changed:
                 if not self._save():
                     # The lease/last_run_at advance was NOT persisted (full
@@ -259,10 +283,10 @@ class FileScheduleStore:
         # Record suppressed occurrences so a missed run is visible in history
         # rather than silently dropped. Logged outside the file lock (log_run
         # takes it itself; flock is not re-entrant across file handles).
-        for job_id, job_name in missed:
+        for job_id, job_name, status in missed:
             self.log_run(
                 job_id=job_id,
-                status="missed",
+                status=status,
                 job_name=job_name,
             )
         return claimed

src/praisonai-agents/praisonaiagents/session/hierarchy.py

@@ -306,6 +306,8 @@ class HierarchicalSessionStore(DefaultSessionStore):
                 raw = f.read()
             data = json.loads(raw.decode("utf-8"))
             session = ExtendedSessionData.from_dict(data)
+            if not self._owns_file(session_id, filepath, session):
+                return ExtendedSessionData(session_id=session_id)
             # Bind the fingerprint to this parsed object and these exact bytes.
             # A later writer must not certify an older payload via its stats.
             with self._lock:

src/praisonai-agents/praisonaiagents/session/sqlite_store.py

@@ -35,7 +35,13 @@ from typing import Any, Dict, List, Optional
 
 from praisonaiagents._logging import get_logger
 
-from .store import DefaultSessionStore, FileLock, SessionData
+from .store import (
+    DefaultSessionStore,
+    FileLock,
+    SessionData,
+    _legacy_session_filename,
+    _safe_session_filename,
+)
 
 logger = get_logger(__name__)
 
@@ -390,6 +396,22 @@ class SqliteSessionStore(DefaultSessionStore):
         except Exception:
             return set()
 
+    def _stored_session_id(self, filename: str) -> Optional[str]:
+        """Return the id stored in ``filename`` if it maps back to that file."""
+        import json
+
+        try:
+            with open(os.path.join(self.session_dir, filename), "r", encoding="utf-8") as f:
+                sid = json.load(f).get("session_id")
+        except (OSError, ValueError, AttributeError):
+            return None
+        if isinstance(sid, str) and filename in (
+            f"{_safe_session_filename(sid)}.json",
+            f"{_legacy_session_filename(sid)}.json",
+        ):
+            return sid
+        return None
+
     def _reindex_all(self) -> bool:
         """Backfill the index from existing JSON transcripts.
 
@@ -433,6 +455,10 @@ class SqliteSessionStore(DefaultSessionStore):
             if not filename.endswith(".json"):
                 continue
             sid = filename[:-5]
+            if "~" in sid or sid not in already:
+                # Hashed or pre-#6024 legacy name (Issue #6024): the real id
+                # lives in the file; loading it migrates a legacy file.
+                sid = self._stored_session_id(filename) or sid
             if sid in pending:
                 # Failed restore retries must not be bypassed by a synthetic
                 # empty transcript from the initial/upgrade backfill path.

src/praisonai-agents/praisonaiagents/session/store.py

@@ -6,6 +6,7 @@ Zero dependencies beyond stdlib.
 """
 
 import copy
+import hashlib
 import json
 from praisonaiagents._logging import get_logger
 import os
@@ -30,6 +31,27 @@ from ..paths import get_sessions_dir
 
 logger = get_logger(__name__)
 
+
+
+def _legacy_session_filename(session_id: str) -> str:
+    """Pre-#6024 lossy filename (many ids -> one name); kept only for migration."""
+    return "".join(c if c.isalnum() or c in "-_" else "_" for c in session_id)
+
+
+def _safe_session_filename(session_id: str) -> str:
+    """Injective, filesystem-safe filename stem for a session id (Issue #6024).
+
+    Ids already made of ``[A-Za-z0-9_-]`` keep their exact legacy name; any
+    other id gets a readable prefix plus a digest of the full id, so ids that
+    differ only in ``.``/``@``/``:``/… never share a file.
+    """
+    readable = _legacy_session_filename(session_id)
+    if readable == session_id:
+        return readable
+    digest = hashlib.sha256(session_id.encode("utf-8")).hexdigest()[:16]
+    return f"{readable[:64]}~{digest}"
+
+
 # Module-level sentinel to track if we've warned about degraded locking
 _WARNED_NO_FCNTL = False
 
@@ -884,10 +906,56 @@ class DefaultSessionStore:
         )
 
     def _get_session_path(self, session_id: str) -> str:
-        """Get the file path for a session."""
-        # Sanitize session_id for filesystem
-        safe_id = "".join(c if c.isalnum() or c in "-_" else "_" for c in session_id)
-        return os.path.join(self.session_dir, f"{safe_id}.json")
+        """Get the file path for a session (injective mapping, Issue #6024)."""
+        path = self._session_file_path(session_id)
+        legacy = os.path.join(self.session_dir, f"{_legacy_session_filename(session_id)}.json")
+        # Migrate a pre-#6024 file only when it is provably this id's own.
+        # Recheck under the destination lock so a racing worker that already
+        # migrated and wrote here is never clobbered by a stale move.
+        if legacy != path and not os.path.exists(path) and os.path.exists(legacy):
+            try:
+                with FileLock(path, self.lock_timeout):
+                    if not os.path.exists(path):
+                        with open(legacy, "r", encoding="utf-8") as f:
+                            owner = json.load(f).get("session_id")
+                        if owner == session_id:
+                            os.replace(legacy, path)
+            except (OSError, ValueError, AttributeError):
+                pass
+        return path
+
+    def _session_file_path(self, session_id: str) -> str:
+        """Session file path without legacy migration (takes no lock)."""
+        return os.path.join(self.session_dir, f"{_safe_session_filename(session_id)}.json")
+
+    def _owns_file(self, session_id: str, filepath: str, session: SessionData) -> bool:
+        """Return False if the loaded file belongs to another session id.
+
+        A foreign file (e.g. a pre-#6024 legacy collision) is never served to
+        the caller; it is moved to its owner's own path (or set aside when that
+        is taken) so a later write here cannot destroy the owner's history.
+        Caller must hold the ``filepath`` lock. Raises ``OSError`` if the file
+        cannot be moved, so writers abort instead of overwriting it.
+        """
+        if not session.session_id:
+            session.session_id = session_id
+            return True
+        if session.session_id == session_id:
+            return True
+        logger.warning(
+            "Session file %s belongs to session %r, not %r; starting fresh",
+            filepath, session.session_id, session_id,
+        )
+        owner_path = self._session_file_path(session.session_id)
+        moved = False
+        if owner_path != filepath:
+            with FileLock(owner_path, self.lock_timeout):
+                if not os.path.exists(owner_path):
+                    os.replace(filepath, owner_path)
+                    moved = True
+        if not moved and self._quarantine_corrupt(filepath) is None:
+            raise OSError(f"cannot move foreign session file {filepath}")
+        return False
     
     def _load_session(self, session_id: str) -> SessionData:
         """Load session from disk with file locking."""
@@ -952,7 +1020,10 @@ class DefaultSessionStore:
         try:
             with open(filepath, "r", encoding="utf-8") as f:
                 data = json.load(f)
-            return SessionData.from_dict(data)
+            session = SessionData.from_dict(data)
+            if not self._owns_file(session_id, filepath, session):
+                return SessionData(session_id=session_id)
+            return session
         except (json.JSONDecodeError, UnicodeDecodeError) as e:
             # Malformed content — the parsed data is unusable, but the raw file
             # may still hold recoverable bytes. Quarantine it aside so the next
@@ -1122,9 +1193,7 @@ class DefaultSessionStore:
         try:
             spill_dir = self._spill_dir()
             os.makedirs(spill_dir, exist_ok=True)
-            safe_id = "".join(
-                c if c.isalnum() or c in "-_" else "_" for c in session_id
-            )
+            safe_id = _safe_session_filename(session_id)
             # A monotonic ms timestamp keeps files sortable/attributable, but a
             # short random token guarantees uniqueness so consecutive failures
             # within the same PID+millisecond never overwrite an earlier spill
@@ -1235,13 +1304,15 @@ class DefaultSessionStore:
             spill_dir = self._spill_dir()
             if not os.path.isdir(spill_dir):
                 return
-            safe_id = "".join(
-                c if c.isalnum() or c in "-_" else "_" for c in session_id
+            # Also match pre-#6024 spill names; the session_id check below
+            # rejects any spill that belongs to a colliding id.
+            prefixes = (
+                f"{_safe_session_filename(session_id)}.",
+                f"{_legacy_session_filename(session_id)}.",
             )
-            prefix = f"{safe_id}."
             candidates = sorted(
                 f for f in os.listdir(spill_dir)
-                if f.startswith(prefix) and f.endswith(".json")
+                if f.startswith(prefixes) and f.endswith(".json")
             )
         except (IOError, OSError):
             return
@@ -1331,7 +1402,9 @@ class DefaultSessionStore:
         new_messages = [m for _, msgs in recovered for m in msgs]
         if new_messages:
             session.messages.extend(new_messages)
-            filepath = self._get_session_path(session_id)
+            # Callers already hold this path's lock; the migrating lookup
+            # would take it again and stall for ``lock_timeout``.
+            filepath = self._session_file_path(session_id)
             session.updated_at = datetime.now(timezone.utc).isoformat()
             # Recovered turns extend the active window like any other append, so
             # they must go through the same retention policy (compact/truncate)

src/praisonai-agents/praisonaiagents/telemetry/telemetry.py

@@ -540,21 +540,21 @@ class MinimalTelemetry:
                 
                 # Use a timeout-based flush to prevent hanging
                 import threading
-                import time
-                import concurrent.futures
-                
-                # Use ThreadPoolExecutor for better control
-                with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
-                    flush_future = executor.submit(self._safe_flush_posthog, posthog_client)
-                    
-                    try:
-                        flush_future.result(timeout=5.0)  # 5 second timeout
-                        self.logger.debug("PostHog flush completed successfully")
-                    except concurrent.futures.TimeoutError:
-                        self.logger.warning("PostHog flush timed out")
-                        flush_future.cancel()
-                    except Exception as e:
-                        self.logger.error(f"PostHog flush failed: {e}")
+
+                # Daemon thread + bounded join: a ThreadPoolExecutor would join
+                # a wedged flush on exit/shutdown, so the timeout never applied.
+                flush_thread = threading.Thread(
+                    target=self._safe_flush_posthog, args=(posthog_client,),
+                    name="praisonai-posthog-flush", daemon=True,
+                )
+                flush_thread.start()
+                flush_thread.join(timeout=5.0)  # 5 second timeout
+                if flush_thread.is_alive():
+                    # Consumer flush/shutdown would block on the same wedged
+                    # transport, so skip the remaining cleanup.
+                    self.logger.warning("PostHog flush timed out")
+                    return
+                self.logger.debug("PostHog flush completed")
                 
                 # Cleanup PostHog threads safely
                 self._shutdown_posthog_threads(posthog_client)

src/praisonai-agents/praisonaiagents/tools/delegation_tools.py

@@ -62,21 +62,22 @@ class DelegationTools:
             # blocking past the documented deadline. A non-positive timeout
             # means "no bound".
             if timeout and timeout > 0:
-                import concurrent.futures
+                # Daemon worker: a ThreadPoolExecutor worker is joined at
+                # interpreter exit, so a hung delegate would keep the process alive.
+                from .call_executor import ToolTimeoutError, _run_with_timeout
 
-                with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
-                    future = pool.submit(
-                        spawn, task=task_description, agent_name=agent_type
+                try:
+                    result = _run_with_timeout(
+                        lambda: spawn(task=task_description, agent_name=agent_type),
+                        timeout,
                     )
-                    try:
-                        result = future.result(timeout=timeout)
-                    except concurrent.futures.TimeoutError:
-                        return json.dumps({
-                            "success": False,
-                            "task_description": task_description,
-                            "agent_type": agent_type,
-                            "error": f"Delegated task timed out after {timeout}s",
-                        }, indent=2)
+                except ToolTimeoutError:
+                    return json.dumps({
+                        "success": False,
+                        "task_description": task_description,
+                        "agent_type": agent_type,
+                        "error": f"Delegated task timed out after {timeout}s",
+                    }, indent=2)
             else:
                 result = spawn(task=task_description, agent_name=agent_type)
 

src/praisonai-agents/praisonaiagents/tools/schedule_tools.py

@@ -148,8 +148,9 @@ def schedule_add(
             default ``PRAISONAI_SCHEDULE_TIMEZONE`` is used, then UTC.
         once: When True the job is a one-shot — it is removed automatically
             after its single successful fire (maps to ``delete_after_run``).
-            Useful for ``at:`` / ``in ...`` reminders so a spent job does not
-            linger in listings.
+            A failed fire leaves it listed but disabled; a run lost to a crash
+            is recorded as ``abandoned`` and retried once. Useful for ``at:`` /
+            ``in ...`` reminders so a spent job does not linger in listings.
         max_runs: Optional run-count bound for a recurring schedule — fire at
             most this many times, then auto-remove (e.g. "remind me 3 times").
             ``0`` (default) means unbounded. Generalises ``once`` (``max_runs=1``).

src/praisonai-agents/praisonaiagents/tools/subagent_tool.py

@@ -66,6 +66,35 @@ def _emit_subagent_stop(
         logger.debug("SUBAGENT_STOP hook failed", exc_info=True)
 
 
+def _messenger_delivery(messenger: Any) -> Callable[[Any], None]:
+    """Build an ``on_complete`` callback that sends a finished job's result
+    back to the turn's own chat (``"origin"``) through its outbound messenger.
+
+    Raises when the send is not accepted so the job manager leaves the job
+    un-``delivered`` instead of recording a delivery that never happened.
+    """
+
+    def _deliver(info: Any) -> None:
+        from .messaging_tools import _run_async
+
+        result = info.result
+        if info.error:
+            text = f"Background task failed: {info.error}"
+        elif isinstance(result, dict):
+            text = result.get("output") if result.get("success") else (
+                f"Background task failed: {result.get('error')}"
+            )
+        else:
+            text = result
+        if not text:
+            text = "Background task finished with no output."
+        outcome = _run_async(messenger.send("origin", str(text)))
+        if not getattr(outcome, "ok", False):
+            raise RuntimeError(getattr(outcome, "summary", "") or "delivery failed")
+
+    return _deliver
+
+
 def create_subagent_tool(
     agent_factory: Optional[Callable[..., Any]] = None,
     allowed_agents: Optional[List[str]] = None,
@@ -114,9 +143,11 @@ def create_subagent_tool(
             ``deliver`` target finishes. This is the observable completion
             signal a gateway (e.g. ``BotOS``) subscribes to in order to
             route the result back to the originating chat without an active
-            turn. When ``None`` (the default) background jobs remain strictly
-            pull-only via ``subagent_result`` — byte-for-byte the prior
-            behaviour.
+            turn. When ``None`` (the default), ``deliver="origin"`` targets
+            the turn's own chat, the spawning turn has an outbound messenger
+            registered (a running gateway) and the send policy allows it, the
+            result is sent back through that messenger; otherwise background
+            jobs remain pull-only via ``subagent_result``.
         
     Returns:
         Tool definition dictionary
@@ -356,16 +387,45 @@ def create_subagent_tool(
             origin: Dict[str, Any] = {}
             job_on_complete = None
             if deliver:
+                # Fill the origin from the current turn (still in scope here)
+                # rather than from the model; explicit arguments still win.
+                from praisonaiagents.session.context import (
+                    get_outbound_messenger,
+                    get_session_context,
+                )
+
+                ctx = get_session_context()
+                use_ctx = not chat_id or chat_id == ctx.chat_id
                 origin = {
                     "deliver": deliver,
-                    "platform": platform,
-                    "chat_id": chat_id,
-                    "thread_id": thread_id,
+                    "platform": platform or (ctx.platform if use_ctx else ""),
+                    "chat_id": chat_id or ctx.chat_id,
+                    "thread_id": thread_id or (ctx.thread_id if use_ctx else ""),
                     "session_id": session_id,
+                    "user_id": ctx.user_id,
                     "task": task,
                 }
                 if on_job_complete is not None:
                     job_on_complete = on_job_complete
+                else:
+                    # Auto-send only back to the turn's own chat and only if
+                    # the send policy allows it (as ``send_message`` does);
+                    # any other target needs an explicit ``on_job_complete``.
+                    from .messaging_tools import _check_send_policy
+
+                    messenger = get_outbound_messenger()
+                    same_chat = (
+                        (not chat_id or chat_id == ctx.chat_id)
+                        and (not platform or platform == ctx.platform)
+                        and (not thread_id or thread_id == ctx.thread_id)
+                    )
+                    if (
+                        messenger is not None
+                        and deliver == "origin"
+                        and same_chat
+                        and _check_send_policy(deliver) is None
+                    ):
+                        job_on_complete = _messenger_delivery(messenger)
 
             job_id = job_manager.start_job(
                 _job,

src/praisonai-agents/praisonaiagents/workflows/yaml_parser.py

@@ -707,7 +707,7 @@ class YAMLWorkflowParser:
         from ..agent import ImageAgent
         
         # ImageAgent accepts: llm, style, and other image config
-        style = config.get('style', 'natural')
+        style = config.get('style')
         
         agent = ImageAgent(
             llm=llm,

src/praisonai-agents/tests/unit/session/test_session_path_collisions.py

@@ -0,0 +1,119 @@
+"""Distinct session ids must never share a session file (Issue #6024)."""
+
+import json
+import os
+
+import pytest
+
+from praisonaiagents.session.store import DefaultSessionStore
+
+
+@pytest.fixture
+def store(tmp_path):
+    return DefaultSessionStore(session_dir=str(tmp_path))
+
+
+@pytest.mark.parametrize("a,b", [
+    ("alice.smith", "alice_smith"),
+    ("a:b", "a_b"),
+    ("[email protected]", "[email protected]"),
+    ("[email protected]", "u_tag_x_com"),
+    ("a/b", "a b"),
+    ("telegram:acct:default:chat:-1001:", "telegram_acct_default_chat_-1001_"),
+])
+def test_colliding_ids_get_distinct_paths(store, a, b):
+    assert store._get_session_path(a) != store._get_session_path(b)
+
+
+def test_safe_ids_keep_legacy_filename(store):
+    sid = "user_123-abc" + "x" * 100
+    assert os.path.basename(store._get_session_path(sid)) == f"{sid}.json"
+
+
+def test_colliding_users_have_separate_histories(store):
+    store.add_user_message("[email protected]", "My PIN is 4471.")
+    assert store.get_chat_history("[email protected]") == []
+    store.add_user_message("[email protected]", "hello")
+    assert [m["content"] for m in store.get_chat_history("[email protected]")] == ["My PIN is 4471."]
+    assert [m["content"] for m in store.get_chat_history("[email protected]")] == ["hello"]
+
+
+def _write_legacy(store, stem, owner, content):
+    path = os.path.join(store.session_dir, f"{stem}.json")
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump({"session_id": owner, "messages": [
+            {"role": "user", "content": content, "timestamp": 1.0}]}, f)
+    return path
+
+
+def test_legacy_file_migrated_only_for_owner(store):
+    legacy = _write_legacy(store, "alice_smith", "alice.smith", "secret")
+    # A colliding unsafe id never adopts it.
+    assert store.get_chat_history("alice:smith") == []
+    assert os.path.exists(legacy)
+    # The real owner migrates it on first read.
+    assert [m["content"] for m in store.get_chat_history("alice.smith")] == ["secret"]
+    assert not os.path.exists(legacy)
+
+
+def test_foreign_file_never_served_and_not_destroyed(store):
+    # Safe id whose filename holds another id's unmigrated legacy transcript.
+    _write_legacy(store, "alice_smith", "alice.smith", "secret")
+    assert store.get_chat_history("alice_smith") == []
+    store.add_user_message("alice_smith", "mine")
+    assert [m["content"] for m in store.get_chat_history("alice_smith")] == ["mine"]
+    # The owner's history was relocated rather than overwritten.
+    assert [m["content"] for m in store.get_chat_history("alice.smith")] == ["secret"]
+
+
+def test_failed_foreign_move_aborts_write(store, monkeypatch, tmp_path):
+    monkeypatch.setattr(store, "_spill_dir", lambda: str(tmp_path / "spill"))
+    store.add_user_message("alice.smith", "new")  # owner path taken
+    legacy = _write_legacy(store, "alice_smith", "alice.smith", "secret")
+    monkeypatch.setattr(store, "_quarantine_corrupt", lambda p: None)
+    assert store.add_user_message("alice_smith", "mine") is False
+    with open(legacy, encoding="utf-8") as f:
+        assert json.load(f)["session_id"] == "alice.smith"
+
+
+def test_foreign_file_set_aside_when_owner_path_taken(store):
+    store.add_user_message("alice.smith", "new")
+    legacy = _write_legacy(store, "alice_smith", "alice.smith", "secret")
+    store.add_user_message("alice_smith", "mine")
+    assert [m["content"] for m in store.get_chat_history("alice_smith")] == ["mine"]
+    assert [m["content"] for m in store.get_chat_history("alice.smith")] == ["new"]
+    aside = [p for p in os.listdir(store.session_dir) if p.startswith("alice_smith.json.corrupt-")]
+    assert len(aside) == 1 and os.path.exists(legacy)
+
+
+def test_sqlite_backfill_indexes_hashed_filenames(tmp_path):
+    from praisonaiagents.session.sqlite_store import SqliteSessionStore
+
+    DefaultSessionStore(session_dir=str(tmp_path)).add_user_message(
+        "[email protected]", "narwhal record")
+    store = SqliteSessionStore(session_dir=str(tmp_path), db_path=":memory:")
+    assert [h.session_id for h in store.search("narwhal")] == ["[email protected]"]
+
+
+def test_sqlite_backfill_indexes_legacy_collision_files(tmp_path):
+    from praisonaiagents.session.sqlite_store import SqliteSessionStore
+
+    legacy = DefaultSessionStore(session_dir=str(tmp_path))
+    _write_legacy(legacy, "alice_smith", "alice.smith", "walrus record")
+    store = SqliteSessionStore(session_dir=str(tmp_path), db_path=":memory:")
+    assert [h.session_id for h in store.search("walrus")] == ["alice.smith"]
+
+
+def test_spill_recovery_does_not_wait_on_own_lock(store, monkeypatch, tmp_path):
+    import time
+
+    spill = tmp_path / "spill"
+    spill.mkdir()
+    monkeypatch.setattr(store, "_spill_dir", lambda: str(spill))
+    # Legacy name of "alice.smith" belongs to another id, so it is never migrated.
+    _write_legacy(store, "alice_smith", "alice:smith", "theirs")
+    (spill / "alice_smith.1.json").write_text(json.dumps({"session_id": "alice.smith", "messages": [
+        {"role": "user", "content": "spilled", "timestamp": 2.0}]}), encoding="utf-8")
+    start = time.monotonic()
+    assert [m["content"] for m in store.get_chat_history("alice.smith")] == ["spilled"]
+    assert time.monotonic() - start < store.lock_timeout

src/praisonai-agents/tests/unit/telemetry/test_shutdown_timeout.py

@@ -0,0 +1,36 @@
+"""Telemetry shutdown must return promptly when the PostHog flush is wedged."""
+
+import threading
+import time
+from unittest.mock import MagicMock
+
+from praisonaiagents.telemetry.telemetry import MinimalTelemetry
+
+
+class _FastJoinThread(threading.Thread):
+    def join(self, timeout=None):
+        return super().join(timeout=0.2 if timeout else timeout)
+
+
+def test_shutdown_returns_when_flush_hangs(monkeypatch):
+    release = threading.Event()
+    client = MagicMock()
+    client.flush.side_effect = lambda: release.wait(10)
+
+    telemetry = MinimalTelemetry(enabled=False)
+    telemetry.enabled = True
+    telemetry._shutdown_lock = threading.Lock()
+    telemetry._shutdown_complete = False
+    telemetry._posthog = client
+    monkeypatch.setattr(telemetry, "flush", lambda: None)
+    monkeypatch.setattr(threading, "Thread", _FastJoinThread)
+
+    start = time.monotonic()
+    try:
+        telemetry.shutdown()
+    finally:
+        release.set()
+    assert time.monotonic() - start < 2
+    # Cleanup that would block on the same wedged transport is skipped.
+    client.shutdown.assert_not_called()
+    assert telemetry._posthog is None

src/praisonai-agents/tests/unit/test_handoff_context_policy.py

@@ -0,0 +1,66 @@
+"""Handoff context policies must keep tool-call pairs intact and keep the goal."""
+
+from types import SimpleNamespace
+
+from praisonaiagents.agent.handoff import (
+    ContextPolicy,
+    Handoff,
+    HandoffConfig,
+    _tool_pair_safe_tail,
+)
+
+CALL = {"id": "c1", "type": "function", "function": {"name": "search", "arguments": "{}"}}
+HISTORY = [
+    {"role": "system", "content": "sys"},
+    {"role": "user", "content": "GOAL: book flight"},
+    {"role": "assistant", "content": "ok, searching"},
+    {"role": "assistant", "content": None, "tool_calls": [CALL]},
+    {"role": "tool", "tool_call_id": "c1", "content": "results"},
+    {"role": "assistant", "content": "found 3 options"},
+]
+
+
+def _prepare(policy, n=10):
+    target = SimpleNamespace(name="tgt")
+    handoff = Handoff(agent=target, config=HandoffConfig(context_policy=policy, max_context_messages=n))
+    source = SimpleNamespace(name="src", chat_history=list(HISTORY))
+    return handoff._prepare_context(source, {}).messages
+
+
+def _assert_pairs_valid(messages):
+    call_ids = {c["id"] for m in messages for c in (m.get("tool_calls") or [])}
+    response_ids = {m["tool_call_id"] for m in messages if m.get("role") == "tool"}
+    assert call_ids == response_ids
+
+
+def test_last_n_drops_orphan_tool_result():
+    msgs = _prepare(ContextPolicy.LAST_N, n=2)
+    _assert_pairs_valid(msgs)
+    assert [m["role"] for m in msgs] == ["system", "assistant"]
+
+
+def test_last_n_keeps_complete_pair():
+    msgs = _prepare(ContextPolicy.LAST_N, n=3)
+    _assert_pairs_valid(msgs)
+    assert any(m.get("tool_calls") for m in msgs)
+
+
+def test_summary_keeps_goal_and_valid_pairs():
+    msgs = _prepare(ContextPolicy.SUMMARY)
+    _assert_pairs_valid(msgs)
+    assert msgs[0]["role"] == "system"
+    assert msgs[1]["content"] == "GOAL: book flight"
+    assert msgs[-1]["content"] == "found 3 options"
+
+
+def test_summary_does_not_duplicate_goal_in_tail():
+    target = SimpleNamespace(name="tgt")
+    handoff = Handoff(agent=target, config=HandoffConfig(context_policy=ContextPolicy.SUMMARY))
+    source = SimpleNamespace(name="src", chat_history=[{"role": "user", "content": "hi"}])
+    assert handoff._prepare_context(source, {}).messages == [{"role": "user", "content": "hi"}]
+
+
+def test_tail_drops_tool_calls_without_responses():
+    msgs = HISTORY[1:4]  # ends on an unanswered tool_calls message
+    assert _tool_pair_safe_tail(msgs, 3) == HISTORY[1:3]
+    assert _tool_pair_safe_tail(msgs, 0) == []

src/praisonai-agents/tests/unit/test_image_agent_style.py

@@ -9,7 +9,9 @@ from praisonaiagents import ImageAgent
     ("openai/dall-e-2", {}, None),
     ("openai/gpt-image-1", {}, None),
     ("openai/gpt-image-1", {"style": "vivid"}, None),
-    ("openai/dall-e-3", {}, "natural"),
+    (None, {}, None),
+    ("openai/dall-e-3", {}, None),
+    ("openai/dall-e-3", {"style": "natural"}, "natural"),
     ("openai/dall-e-3", {"style": "vivid"}, "vivid"),
 ])
 def test_style_forwarding(llm, kwargs, expected):
@@ -29,3 +31,17 @@ def test_style_forwarding(llm, kwargs, expected):
             assert received["style"] == expected
     finally:
         agent.close()
+
+
+@pytest.mark.parametrize("config,expected", [
+    ({}, None),
+    ({"style": "vivid"}, "vivid"),
+])
+def test_yaml_image_agent_style(config, expected):
+    from praisonaiagents.workflows.yaml_parser import YAMLWorkflowParser
+
+    agent = YAMLWorkflowParser()._create_image_agent("img", "openai/dall-e-3", config)
+    try:
+        assert agent.image_config.style == expected
+    finally:
+        agent.close()

src/praisonai-agents/tests/unit/test_schedule_claim.py

@@ -7,7 +7,7 @@ Tests:
 - Concurrent claims (threads) each get disjoint jobs — no double-fire
 - complete() releases the lease
 - Expired leases become claimable again (crash recovery)
-- One-shot jobs (delete_after_run) are removed on claim
+- One-shot jobs (delete_after_run) stay leased on claim, retired on completion
 - Lease metadata round-trips through to_dict/from_dict
 - ScheduleRunner.claim_due_jobs falls back to get_due_jobs without claim
 """
@@ -78,7 +78,7 @@ class TestClaimDue:
             reclaimed = store.claim_due(t1, owner_id="B", lease_seconds=5)
             assert len(reclaimed) == 1
 
-    def test_one_shot_removed_on_claim(self):
+    def test_one_shot_retired_on_completion(self):
         with tempfile.TemporaryDirectory() as d:
             store = FileScheduleStore(store_dir=d)
             job = ScheduleJob(
@@ -90,6 +90,10 @@ class TestClaimDue:
             store.add(job)
             claimed = store.claim_due(time.time(), owner_id="A")
             assert len(claimed) == 1
+            # Kept (leased, retiring) so a crash mid-run cannot lose it.
+            kept = store.get(job.id)
+            assert kept is not None and kept._retiring == 1
+            ScheduleRunner(store).mark_run(claimed[0])
             assert store.get(job.id) is None
 
     def test_concurrent_claims_no_double_fire(self):
@@ -335,7 +339,7 @@ class TestConfigYamlClaimDue:
             reclaimed = store.claim_due(t1, owner_id="B", lease_seconds=5)
             assert len(reclaimed) == 1
 
-    def test_one_shot_removed_on_claim(self):
+    def test_one_shot_retired_on_completion(self):
         with tempfile.TemporaryDirectory() as d:
             store = self._make_store(d)
             job = ScheduleJob(
@@ -347,6 +351,10 @@ class TestConfigYamlClaimDue:
             store.add(job)
             claimed = store.claim_due(time.time(), owner_id="A")
             assert len(claimed) == 1
+            # Kept (leased, retiring) so a crash mid-run cannot lose it.
+            kept = store.get(job.id)
+            assert kept is not None and kept._retiring == 1
+            ScheduleRunner(store).mark_run(claimed[0])
             assert store.get(job.id) is None
 
     def test_concurrent_claims_no_double_fire(self):

src/praisonai-agents/tests/unit/test_schedule_crash_recovery.py

@@ -0,0 +1,183 @@
+"""
+Crash recovery for claimed schedule jobs (issue #6036).
+
+A spent job (one-shot / ``max_runs`` met) is no longer deleted at claim time:
+it stays leased and is retired on completion. A lease that expires without
+``complete`` is recorded as an ``abandoned`` run; a one-shot is retried once,
+a recurring job waits for its next slot. Covers both stores.
+"""
+
+import os
+import time
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from praisonaiagents.scheduler.config_store import ConfigYamlScheduleStore
+from praisonaiagents.scheduler.models import Schedule, ScheduleJob
+from praisonaiagents.scheduler.runner import ScheduleRunner
+from praisonaiagents.scheduler.store import FileScheduleStore
+
+
+@pytest.fixture(params=["file", "yaml"])
+def make_store(request, tmp_path):
+    if request.param == "file":
+        return lambda: FileScheduleStore(store_dir=str(tmp_path))
+    path = os.path.join(str(tmp_path), "config.yaml")
+    return lambda: ConfigYamlScheduleStore(config_path=path)
+
+
+def _one_shot(name="remind-once"):
+    past = (datetime.now(timezone.utc) - timedelta(seconds=5)).isoformat()
+    return ScheduleJob(
+        name=name,
+        message="ping",
+        schedule=Schedule(kind="at", at=past),
+        delete_after_run=True,
+    )
+
+
+def _statuses(store, job_id):
+    return [r.status for r in store.get_history(job_id=job_id)]
+
+
+def test_claim_keeps_one_shot_until_completion(make_store):
+    store = make_store()
+    job = _one_shot()
+    store.add(job)
+    claimed = store.claim_due(time.time(), owner_id="A", lease_seconds=300)
+    assert [j.id for j in claimed] == [job.id]
+    kept = make_store().get(job.id)
+    assert kept is not None and kept._retiring == 1
+    assert kept._lease_owner == "A"
+    # A competitor cannot claim it while leased.
+    assert make_store().claim_due(time.time(), owner_id="B") == []
+    ScheduleRunner(store).mark_run(claimed[0])
+    store.complete(job.id, "A")
+    assert make_store().get(job.id) is None
+
+
+def test_crashed_one_shot_is_retried_once(make_store):
+    store = make_store()
+    job = _one_shot()
+    store.add(job)
+    t0 = time.time()
+    assert store.claim_due(t0, owner_id="A", lease_seconds=1)
+    del store  # crash: no mark_run / complete
+
+    restarted = make_store()
+    retry = restarted.claim_due(t0 + 5, owner_id="B", lease_seconds=1)
+    assert [j.id for j in retry] == [job.id]
+    assert _statuses(restarted, job.id) == ["abandoned"]
+    # Retry also crashes: logged and retired, never offered a third time.
+    third = make_store()
+    assert third.claim_due(t0 + 10, owner_id="C", lease_seconds=1) == []
+    assert third.get(job.id) is None
+    assert _statuses(make_store(), job.id) == ["abandoned", "abandoned"]
+
+
+def test_retried_one_shot_completes_once(make_store):
+    store = make_store()
+    job = _one_shot()
+    store.add(job)
+    t0 = time.time()
+    store.claim_due(t0, owner_id="A", lease_seconds=1)
+    restarted = make_store()
+    retry = restarted.claim_due(t0 + 5, owner_id="B", lease_seconds=300)
+    ScheduleRunner(restarted).mark_run(retry[0])
+    restarted.complete(job.id, "B")
+    final = make_store()
+    assert final.get(job.id) is None
+    assert sorted(_statuses(final, job.id)) == ["abandoned", "succeeded"]
+
+
+def test_crashed_recurring_job_logs_abandoned_and_waits_for_next_slot(make_store):
+    store = make_store()
+    job = ScheduleJob(
+        name="hourly", message="digest",
+        schedule=Schedule(kind="every", every_seconds=3600),
+    )
+    store.add(job)
+    t0 = time.time()
+    assert store.claim_due(t0, owner_id="A", lease_seconds=1)
+
+    restarted = make_store()
+    assert restarted.claim_due(t0 + 5, owner_id="B") == []
+    assert _statuses(restarted, job.id) == ["abandoned"]
+    assert restarted.get(job.id) is not None
+    # Next slot fires normally.
+    assert len(restarted.claim_due(t0 + 3601, owner_id="B")) == 1
+
+
+def test_failed_one_shot_stays_listed_disabled(make_store):
+    store = make_store()
+    job = _one_shot()
+    store.add(job)
+    claimed = store.claim_due(time.time(), owner_id="A")
+    ScheduleRunner(store).mark_run(claimed[0], status="failed", error="provider 503")
+    store.complete(job.id, "A")
+    after = make_store()
+    kept = after.get(job.id)
+    assert kept is not None and kept.enabled is False
+    assert _statuses(after, job.id) == ["failed"]
+    # Disabled: never re-claimed or silently retired.
+    assert after.claim_due(time.time() + 10, owner_id="B") == []
+    assert after.get(job.id) is not None
+
+
+def test_one_shot_in_flight_past_lease_not_refired_by_same_process(make_store):
+    store = make_store()
+    job = _one_shot()
+    store.add(job)
+    t0 = time.time()
+    assert store.claim_due(t0, owner_id="A", lease_seconds=1)
+    # Same ticker polls again while its run is still going.
+    assert store.claim_due(t0 + 5, owner_id="A", lease_seconds=1) == []
+
+
+def test_future_one_shot_survives_poll_before_its_time(make_store):
+    store = make_store()
+    future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
+    job = ScheduleJob(
+        name="later", message="ping",
+        schedule=Schedule(kind="at", at=future), delete_after_run=True,
+    )
+    store.add(job)
+    assert store.claim_due(time.time(), owner_id="A") == []
+    assert make_store().get(job.id) is not None
+
+
+def test_resumed_failed_one_shot_runs_again(make_store):
+    store = make_store()
+    job = _one_shot()
+    store.add(job)
+    claimed = store.claim_due(time.time(), owner_id="A")
+    ScheduleRunner(store).mark_run(claimed[0], status="failed", error="boom")
+    store.complete(job.id, "A")
+    resumed = make_store()
+    kept = resumed.get(job.id)
+    kept.enabled = True
+    resumed.update(kept)
+    retry = resumed.claim_due(time.time() + 10, owner_id="B")
+    assert [j.id for j in retry] == [job.id]
+    ScheduleRunner(resumed).mark_run(retry[0])
+    resumed.complete(job.id, "B")
+    assert make_store().get(job.id) is None
+
+
+def test_crashed_spent_job_not_retried_past_until(make_store):
+    store = make_store()
+    t0 = time.time()
+    until = datetime.fromtimestamp(t0 + 3, timezone.utc).isoformat()
+    job = ScheduleJob(
+        name="bounded", message="ping",
+        schedule=Schedule(kind="every", every_seconds=3600),
+        max_runs=1, until=until,
+    )
+    store.add(job)
+    assert store.claim_due(t0, owner_id="A", lease_seconds=1)
+    # Crash, then restart after the deadline: logged and retired, not re-run.
+    restarted = make_store()
+    assert restarted.claim_due(t0 + 5, owner_id="B") == []
+    assert restarted.get(job.id) is None
+    assert _statuses(make_store(), job.id) == ["abandoned"]

src/praisonai-agents/tests/unit/test_schedule_lifecycle_hooks.py

@@ -241,9 +241,9 @@ class TestFileScheduleStoreEmits:
 
 
 class TestOneShotAutoDelete:
-    """A spent ``delete_after_run`` job is deleted inside ``claim_due``,
-    bypassing ``remove()`` entirely — it must still announce itself, exactly
-    once (the runner's later ``remove(job.id)`` finds nothing)."""
+    """A spent ``delete_after_run`` job is kept (leased) by ``claim_due`` and
+    retired by the runner's ``mark_run`` once the run completes — announcing
+    itself exactly once."""
 
     def test_claim_due_auto_delete_fires_once(self, tmp_path):
         from praisonaiagents.scheduler.store import FileScheduleStore
@@ -258,12 +258,15 @@ class TestOneShotAutoDelete:
         with _Recorder(HookEvent.SCHEDULE_REMOVE) as rec:
             claimed = store.claim_due(now=time.time() + 5, owner_id="w1")
             assert [j.id for j in claimed] == [job.id]
-            # The runner's post-run cleanup must not double-announce it.
-            assert store.remove(job.id) is False
+            # Not deleted at claim: a crash mid-run must not lose it.
+            assert rec.fired == []
+            ScheduleRunner(store).mark_run(claimed[0])
+            store.complete(job.id, "w1")
+            # A later poll finds nothing left to retire.
+            assert store.claim_due(now=time.time() + 10, owner_id="w1") == []
 
         assert rec.fired == ["schedule_remove"]
         assert rec.payloads[0]["job_name"] == "one-shot"
-        assert ScheduleRunner is not None  # cleanup path lives here
 
     def test_recurring_claim_does_not_fire(self, tmp_path):
         from praisonaiagents.scheduler.store import FileScheduleStore

src/praisonai-agents/tests/unit/tools/test_delegation_tools.py

@@ -101,15 +101,19 @@ def test_delegate_task_timeout_exceeded(monkeypatch):
 
     def factory(*, agent_factory=None, **kwargs):
         def spawn(task, agent_name=None, **kw):
-            time.sleep(2)
+            time.sleep(4)
             return {"success": True, "output": "too late", "agent_name": agent_name}
         return {"function": spawn}
 
     monkeypatch.setattr(delegation_tools, "create_subagent_tool", factory)
+    start = time.monotonic()
     raw = DelegationTools().delegate_task("Slow task", agent_type="research", timeout=1)
+    elapsed = time.monotonic() - start
     data = json.loads(raw)
     assert data["success"] is False
     assert "timed out" in data["error"]
+    # The caller must not wait for the abandoned worker to finish.
+    assert elapsed < 3
 
 
 def test_delegate_task_requires_approval_registered():

src/praisonai-agents/tests/unit/tools/test_subagent_origin_context.py

@@ -0,0 +1,178 @@
+"""spawn_subagent(background=True, deliver=...) inside a gateway turn (#6037).
+
+The model cannot supply the origin chat, so it must come from the turn's
+session context, and the result must reach the turn's outbound messenger.
+"""
+
+import time
+
+from praisonaiagents.background.job_manager import get_job_manager
+from praisonaiagents.gateway.protocols import DeliveryResult
+from praisonaiagents.session.context import (
+    clear_outbound_messenger,
+    clear_session_context,
+    register_outbound_messenger,
+    set_session_context,
+)
+from praisonaiagents.tools.subagent_tool import create_subagent_tool
+
+
+class _StubAgent:
+    def chat(self, prompt):
+        return "long job done"
+
+
+class _FakeMessenger:
+    def __init__(self, ok=True):
+        self.ok = ok
+        self.sent = []
+
+    async def send(self, target, text, *, media=None):
+        self.sent.append((target, text))
+        return DeliveryResult(ok=self.ok, target=target)
+
+    def list_targets(self):
+        return []
+
+
+def _spawn_in_turn(messenger, deliver="origin", **kwargs):
+    tool = create_subagent_tool(agent_factory=lambda **kw: _StubAgent())
+    ctx = set_session_context(
+        platform="telegram", chat_id="100", thread_id="7", user_id="alice"
+    )
+    msg = register_outbound_messenger(messenger)
+    try:
+        return tool["function"](
+            task="summarise", background=True, deliver=deliver, **kwargs
+        )
+    finally:
+        clear_outbound_messenger(msg)
+        clear_session_context(ctx)
+
+
+def _wait_info(job_id):
+    jm = get_job_manager()
+    jm.get_result(job_id, timeout=5)
+    for _ in range(50):
+        info = jm.get_job_info(job_id)
+        if info.delivered:
+            break
+        time.sleep(0.02)
+    return info
+
+
+def test_origin_filled_from_session_context_and_delivered_once():
+    messenger = _FakeMessenger()
+    handle = _spawn_in_turn(messenger)
+    info = _wait_info(handle["job_id"])
+
+    assert info.origin["platform"] == "telegram"
+    assert info.origin["chat_id"] == "100"
+    assert info.origin["thread_id"] == "7"
+    assert info.origin["user_id"] == "alice"
+    assert info.delivered is True
+    assert len(messenger.sent) == 1
+    target, text = messenger.sent[0]
+    assert target == "origin"
+    assert "long job done" in text
+
+
+def test_explicit_origin_arguments_win():
+    messenger = _FakeMessenger()
+    handle = _spawn_in_turn(messenger, platform="slack", chat_id="C9")
+    info = _wait_info(handle["job_id"])
+    assert info.origin["platform"] == "slack"
+    assert info.origin["chat_id"] == "C9"
+    assert info.origin["thread_id"] == ""
+    assert info.origin["user_id"] == "alice"
+    # A different chat is not the turn's chat: never send it to chat "100".
+    assert messenger.sent == []
+    assert info.delivered is False
+
+
+def test_explicit_other_thread_is_not_auto_sent():
+    messenger = _FakeMessenger()
+    handle = _spawn_in_turn(messenger, thread_id="99")
+    info = _wait_info(handle["job_id"])
+    assert info.origin["thread_id"] == "99"
+    assert messenger.sent == []
+    assert info.delivered is False
+
+
+def test_non_origin_target_is_not_auto_sent():
+    messenger = _FakeMessenger()
+    handle = _spawn_in_turn(messenger, deliver="telegram:999")
+    info = _wait_info(handle["job_id"])
+    assert messenger.sent == []
+    assert info.delivered is False
+
+
+def test_send_policy_denial_blocks_auto_send():
+    from praisonaiagents.gateway.protocols import SendDecision
+    from praisonaiagents.session.context import (
+        clear_send_policy,
+        register_send_policy,
+    )
+
+    class _Deny:
+        def evaluate(self, target, **kw):
+            return SendDecision(allow=False, reason="no")
+
+    messenger = _FakeMessenger()
+    token = register_send_policy(_Deny())
+    try:
+        handle = _spawn_in_turn(messenger)
+    finally:
+        clear_send_policy(token)
+    info = _wait_info(handle["job_id"])
+    assert messenger.sent == []
+    assert info.delivered is False
+
+
+def test_rejected_send_is_not_marked_delivered():
+    messenger = _FakeMessenger(ok=False)
+    handle = _spawn_in_turn(messenger)
+    jm = get_job_manager()
+    jm.get_result(handle["job_id"], timeout=5)
+    time.sleep(0.1)
+    assert len(messenger.sent) == 1
+    assert jm.get_job_info(handle["job_id"]).delivered is False
+
+
+def test_without_gateway_stays_pull_only():
+    tool = create_subagent_tool(agent_factory=lambda **kw: _StubAgent())
+    handle = tool["function"](task="t", background=True, deliver="origin")
+    jm = get_job_manager()
+    jm.get_result(handle["job_id"], timeout=5)
+    time.sleep(0.05)
+    info = jm.get_job_info(handle["job_id"])
+    assert info.delivered is False
+    assert info.origin["chat_id"] == ""
+
+
+def test_restating_own_chat_keeps_turn_thread_and_platform():
+    messenger = _FakeMessenger()
+    handle = _spawn_in_turn(messenger, chat_id="100")
+    info = _wait_info(handle["job_id"])
+    assert info.origin["platform"] == "telegram"
+    assert info.origin["thread_id"] == "7"
+    assert len(messenger.sent) == 1
+
+
+def test_empty_output_still_notifies_before_marking_delivered():
+    class _Silent:
+        def chat(self, prompt):
+            return ""
+
+    tool = create_subagent_tool(agent_factory=lambda **kw: _Silent())
+    messenger = _FakeMessenger()
+    ctx = set_session_context(platform="telegram", chat_id="100", user_id="alice")
+    msg = register_outbound_messenger(messenger)
+    try:
+        handle = tool["function"](task="t", background=True, deliver="origin")
+    finally:
+        clear_outbound_messenger(msg)
+        clear_session_context(ctx)
+    info = _wait_info(handle["job_id"])
+    assert info.delivered is True
+    assert len(messenger.sent) == 1

src/praisonai-bot/praisonai_bot/bots/_commands.py

@@ -1475,6 +1475,7 @@ def handle_tasks_command(
     user_id: str,
     args: Optional[str] = None,
     runner: Optional[Any] = None,
+    job_manager: Optional[Any] = None,
 ) -> str:
     """Handle /tasks to inspect background tasks from inside a chat session.
 
@@ -1492,6 +1493,9 @@ def handle_tasks_command(
         user_id: The requesting user's identifier (owner scope).
         args: Optional argument string (an id, or ``cancel <id>``).
         runner: Optional ``BackgroundRunner`` (defaults to the shared one).
+        job_manager: Optional ``BackgroundJobManager`` holding background
+            subagents (defaults to the shared one). Jobs are owned through
+            ``origin["user_id"]``, captured when they were spawned.
 
     Returns:
         A compact, chat-friendly text response.
@@ -1504,6 +1508,28 @@ def handle_tasks_command(
         except Exception as e:  # noqa: BLE001
             return f"❌ Background tasks unavailable: {e}"
 
+    if job_manager is None:
+        try:
+            from praisonaiagents.background.job_manager import get_job_manager
+
+            job_manager = get_job_manager()
+        except Exception:  # noqa: BLE001
+            job_manager = None
+
+    def _mine(info: Any) -> bool:
+        # Same fail-closed rule as ``_owned``: jobs without an owner are hidden.
+        owner = (info.origin or {}).get("user_id")
+        return bool(owner) and str(owner) == str(user_id)
+
+    def _owned_job(job_id: str) -> Optional[Any]:
+        if job_manager is None:
+            return None
+        try:
+            info = job_manager.get_job_info(job_id)
+        except Exception:  # noqa: BLE001
+            return None
+        return info if _mine(info) else None
+
     def _owned(task: Dict[str, Any]) -> bool:
         # Fail closed: a task is only visible/actionable to the requesting user
         # when its owner id matches exactly. Ownerless tasks (e.g. submitted
@@ -1525,7 +1551,11 @@ def handle_tasks_command(
         except Exception:  # noqa: BLE001
             task = None
         if task is None or not _owned(task.to_dict()):
-            return f"❌ Task not found: {task_id}"
+            if _owned_job(task_id) is None:
+                return f"❌ Task not found: {task_id}"
+            if not job_manager.cancel_job(task_id):
+                return f"❌ Could not cancel task {task_id} (already finished?)."
+            return f"✅ Cancelled task {task_id}."
         try:
             # Cancel through the runner so the underlying future is actually
             # stopped (not just the record marked cancelled). cancel_task_sync
@@ -1549,7 +1579,21 @@ def handle_tasks_command(
         except Exception:  # noqa: BLE001
             task = None
         if task is None or not _owned(task.to_dict()):
-            return f"❌ Task not found: {arg}"
+            info = _owned_job(arg)
+            if info is None:
+                return f"❌ Task not found: {arg}"
+            lines = [
+                f"🧩 Task {info.job_id} — {info.origin.get('task') or 'subagent'}",
+                f"Status: {info.status.value}",
+            ]
+            if info.error:
+                lines.append(f"Error: {str(info.error)[:200]}")
+            elif info.result is not None:
+                result = info.result
+                if isinstance(result, dict):
+                    result = result.get("output") or result.get("error")
+                lines.append(f"Result: {str(result)[:200]}")
+            return "\n".join(lines)
         t = task.to_dict()
         lines = [
             f"🧩 Task {t.get('id')} — {t.get('name') or 'unnamed'}",
@@ -1566,8 +1610,19 @@ def handle_tasks_command(
         tasks = [t for t in runner.list_tasks() if _owned(t)]
     except Exception as e:  # noqa: BLE001
         return f"❌ Could not list tasks: {e}"
+    try:
+        jobs = sorted(
+            (
+                j for j in (job_manager.list_jobs().values() if job_manager else ())
+                if _mine(j)
+            ),
+            key=lambda j: j.created_at,
+            reverse=True,
+        )
+    except Exception:  # noqa: BLE001
+        jobs = []
 
-    if not tasks:
+    if not tasks and not jobs:
         return "💤 No background tasks."
 
     lines = ["🧩 Background tasks:"]
@@ -1577,6 +1632,9 @@ def handle_tasks_command(
             f"• {t.get('id')} {t.get('name') or 'unnamed'} — "
             f"{t.get('status')} ({progress})"
         )
+    for j in jobs[: max(0, 20 - len(tasks))]:
+        name = str(j.origin.get("task") or "subagent")[:60]
+        lines.append(f"• {j.job_id} {name} — {j.status.value}")
     lines.append("\nUse /tasks <id> for detail or /tasks cancel <id>.")
     return "\n".join(lines)
 

src/praisonai-bot/praisonai_bot/bots/_session.py

@@ -1325,8 +1325,14 @@ class BotSessionManager:
                 from praisonaiagents.session.context import (
                     register_outbound_messenger as _register_messenger,
                     clear_outbound_messenger as _clear_messenger,
+                    register_gateway_loop,
                 )
 
+                # Background workers (e.g. a ``deliver``-backed subagent)
+                # send after the turn ends; record the loop the router's
+                # adapters live on so they schedule onto it.
+                register_gateway_loop(asyncio.get_running_loop())
+
                 origin_source = None
                 if self._platform and chat_id:
                     origin_source = SessionSource(

src/praisonai-bot/praisonai_bot/cli/commands/gateway.py

@@ -1761,7 +1761,7 @@ def gateway_channels_update(
     if token is not None:
         settings["token"] = token
     if not settings:
-        print("Error: nothing to update; pass --token and/or --set key=value")
+        print("Error: nothing to update; pass --set key=value, --token, or both")
         raise typer.Exit(1)
     _edit_channel(ctx, config, name, settings)
     print(f"Updated channel {name!r}.")

src/praisonai-bot/praisonai_bot/scheduler/_standalone_sender.py

@@ -377,10 +377,25 @@ _STANDALONE_SENDERS: dict = {
 def resolve_standalone_sender(channel: str) -> Optional[StandaloneSender]:
     """Return the standalone sender for ``channel``, or ``None`` if none exists.
 
+    Bundled platforms are checked first; otherwise a ``standalone_sender``
+    declared on the channel's registered ``ChannelDescriptor`` is used.
+
     ``None`` means the executor has no out-of-process delivery path for that
     platform and falls back to logging a warning (its prior behaviour when no
     live adapter is wired), so an unsupported platform never raises here.
     """
     if not channel:
         return None
-    return _STANDALONE_SENDERS.get(channel.lower())
+    sender = _STANDALONE_SENDERS.get(channel.lower())
+    if sender is not None:
+        return sender
+    # A ``praisonai.channels`` plugin may declare its own stateless sender on
+    # its ``ChannelDescriptor`` so gateway-less delivery works for it too.
+    try:
+        from praisonai_bot.bots._registry import get_platform_descriptor
+
+        declared = getattr(get_platform_descriptor(channel), "standalone_sender", None)
+    except Exception as e:  # a broken plugin must not crash the tick
+        logger.debug("descriptor lookup failed for channel %r: %s", channel, e)
+        return None
+    return declared if callable(declared) else None

src/praisonai-bot/praisonai_bot/scheduler/executor.py

@@ -1217,11 +1217,11 @@ class ScheduledAgentExecutor:
         prior state untouched); an empty dict likewise changes nothing. No-op
         when the store has no state support.
 
-        One-shot jobs (``delete_after_run``) are already removed from the store
-        by ``claim_due`` before the run reaches here, and their state is popped
-        with them. Persisting now would recreate an orphan ``job_state`` entry
-        that nothing ever cleans up (the schedule is gone, so ``remove`` is a
-        no-op) — and which would be injected if the same explicit id is reused.
+        One-shot jobs (``delete_after_run``) are retired from the store once
+        their run completes, and their state is popped with them. Persisting
+        would recreate an orphan ``job_state`` entry that nothing ever cleans
+        up (the schedule is gone, so ``remove`` is a no-op) — and which
+        would be injected if the same explicit id is reused.
         A single-run job has no "next run" to read a watermark, so skip it.
         """
         if getattr(job, "delete_after_run", False):
@@ -1654,7 +1654,9 @@ class ScheduledAgentExecutor:
                 "no live adapter and no standalone sender for channel "
                 f"{getattr(delivery, 'channel', '')!r}"
             )
-        result = await sender(delivery, text)
+        result = sender(delivery, text)
+        if inspect.isawaitable(result):  # plugin senders may be sync (#6099)
+            result = await result
         return result is not False
 
     def _incident_tracker(self) -> Optional[Any]:

src/praisonai-bot/tests/unit/bots/test_tasks_command_job_manager.py

@@ -0,0 +1,75 @@
+"""/tasks over BackgroundJobManager jobs (background subagents, #6037)."""
+
+import threading
+
+from praisonaiagents.background import BackgroundRunner
+from praisonaiagents.background.job_manager import BackgroundJobManager
+
+from praisonai_bot.bots._commands import handle_tasks_command
+
+
+def _start(jm, owner, release):
+    return jm.start_job(
+        release.wait,
+        origin={"deliver": "origin", "user_id": owner, "task": "research repo"},
+    )
+
+
+def test_list_detail_and_cancel_own_jobs():
+    jm = BackgroundJobManager()
+    release = threading.Event()
+    try:
+        mine = _start(jm, "alice", release)
+        theirs = _start(jm, "bob", release)
+        runner = BackgroundRunner()
+
+        listing = handle_tasks_command("alice", None, runner=runner, job_manager=jm)
+        assert mine in listing
+        assert "research repo" in listing
+        assert theirs not in listing
+
+        detail = handle_tasks_command("alice", mine, runner=runner, job_manager=jm)
+        assert mine in detail
+
+        assert "not found" in handle_tasks_command(
+            "alice", theirs, runner=runner, job_manager=jm
+        )
+        assert "not found" in handle_tasks_command(
+            "alice", f"cancel {theirs}", runner=runner, job_manager=jm
+        )
+
+        out = handle_tasks_command("alice", f"cancel {mine}", runner=runner, job_manager=jm)
+        assert out.startswith("✅")
+        assert jm.get_job_info(mine).status.value == "cancelled"
+        assert jm.get_job_info(theirs).status.value == "running"
+    finally:
+        release.set()
+        jm.shutdown()
+
+
+def test_ownerless_jobs_hidden():
+    jm = BackgroundJobManager()
+    try:
+        job = jm.start_job(lambda: "x")
+        jm.get_result(job, timeout=5)
+        out = handle_tasks_command("alice", None, runner=BackgroundRunner(), job_manager=jm)
+        assert out == "💤 No background tasks."
+    finally:
+        jm.shutdown()
+
+
+def test_newest_jobs_listed_first_when_over_limit():
+    jm = BackgroundJobManager()
+    try:
+        ids = []
+        for _ in range(25):
+            job = jm.start_job(
+                lambda: "x", origin={"deliver": "origin", "user_id": "alice"}
+            )
+            jm.get_result(job, timeout=5)
+            ids.append(job)
+        out = handle_tasks_command("alice", None, runner=BackgroundRunner(), job_manager=jm)
+        assert ids[-1] in out
+        assert ids[0] not in out
+    finally:
+        jm.shutdown()

src/praisonai-bot/tests/unit/scheduler/test_standalone_sender.py

@@ -76,6 +76,52 @@ def test_resolver_known_and_unknown():
     assert ss.resolve_standalone_sender(None) is None  # type: ignore[arg-type]
 
 
+def test_plugin_descriptor_sender_used_when_no_live_handler(monkeypatch):
+    """A ``praisonai.channels`` plugin's declared sender works gateway-less (#6099)."""
+    from praisonai_bot.bots import _registry
+
+    sent: list = []
+
+    class _MatrixDescriptor:
+        config_fields: list = []
+        system_prompt_hint = ""
+
+        async def standalone_sender(self, target, text):
+            sent.append((target.channel_id, text))
+
+    monkeypatch.setattr(
+        _registry, "get_platform_descriptor",
+        lambda name: _MatrixDescriptor() if name == "matrix" else None,
+    )
+    assert ss.resolve_standalone_sender("irc") is None
+
+    ex = _agent_executor(delivery_handler=None)
+    result = _run(ex._execute_one(_job(deliver="matrix:#room")))
+
+    assert result.delivered is True
+    assert result.delivery_error is None
+    assert sent == [("#room", "echo:hello")]
+
+
+def test_plugin_descriptor_sync_sender_supported(monkeypatch):
+    """A plugin may declare a plain (non-async) sender."""
+    from praisonai_bot.bots import _registry
+
+    sent: list = []
+
+    class _SyncDescriptor:
+        def standalone_sender(self, target, text):
+            sent.append(text)
+
+    monkeypatch.setattr(_registry, "get_platform_descriptor", lambda name: _SyncDescriptor())
+
+    ex = _agent_executor(delivery_handler=None)
+    result = _run(ex._execute_one(_job(deliver="matrix:#room")))
+
+    assert result.delivered is True
+    assert sent == ["echo:hello"]
+
+
 # ── executor fallback ────────────────────────────────────────────────
 
 

src/praisonai-bot/tests/unit/test_bot_session_key_isolation.py

@@ -0,0 +1,32 @@
+"""Colliding bot user ids must get separate persisted histories (Issue #6024)."""
+
+import asyncio
+
+from praisonaiagents.session.store import DefaultSessionStore
+from praisonai_bot.bots._session import BotSessionManager
+
+
+class _EchoAgent:
+    def __init__(self):
+        self.name = "support"
+        self.chat_history = []
+        self.seen = {}
+
+    def chat(self, prompt):
+        self.seen[prompt] = [m["content"] for m in self.chat_history if m["role"] == "user"]
+        self.chat_history.append({"role": "user", "content": prompt})
+        self.chat_history.append({"role": "assistant", "content": "ok"})
+        return "ok"
+
+
+def test_colliding_email_senders_are_isolated(tmp_path):
+    mgr = BotSessionManager(store=DefaultSessionStore(session_dir=str(tmp_path)), platform="email")
+    agent = _EchoAgent()
+
+    async def main():
+        await mgr.chat(agent, "[email protected]", "My account PIN is 4471.")
+        await mgr.chat(agent, "[email protected]", "What do you know about me?")
+
+    asyncio.run(main())
+    assert "My account PIN is 4471." not in agent.seen["What do you know about me?"]
+    assert len([p for p in tmp_path.iterdir() if p.suffix == ".json"]) == 2

src/praisonai-code/praisonai_code/cli/commands/session.py

@@ -552,18 +552,15 @@ def session_fork(
     # searched (project first, then global) so the fork source and the
     # existence check stay consistent.
     #
-    # The store persists each session under a *sanitized* filename
-    # (``DefaultSessionStore._get_session_path`` replaces any char that is not
-    # alphanumeric/``-``/``_`` with ``_``). Sanitize identically here so a
-    # global-only id containing e.g. ``.`` or ``:`` still matches its real file
-    # instead of falling through to an empty project-scoped fork.
-    safe_id = "".join(c if c.isalnum() or c in "-_" else "_" for c in session_id)
+    # Ask each store directly: it resolves the injective filename and only
+    # migrates a pre-#6024 legacy file that really belongs to this id
+    # (Issue #6024), so a colliding id's file never selects the wrong store.
     store_dir = str(get_project_sessions_dir())
     for candidate in canonical_cli_stores():
         candidate_dir = getattr(candidate, "session_dir", None)
         if not candidate_dir:
             continue
-        if (Path(candidate_dir) / f"{safe_id}.json").exists():
+        if candidate.session_exists(session_id):
             store_dir = str(candidate_dir)
             break
 

src/praisonai-code/tests/unit/test_session_fork.py

@@ -163,3 +163,23 @@ def test_cli_session_fork_resolves_sanitized_global_id(project):
     # Real history copied, proving the fork resolved the global store (not an
     # empty project-scoped fallback) despite the special characters in the id.
     assert len(forked.messages) == 2
+
+
+def test_cli_session_fork_ignores_colliding_project_file(project):
+    # Issue #6024: a project session ``sess_global`` shares the pre-#6024 lossy
+    # filename with global ``sess.global``; fork must still pick the global one.
+    from praisonai_code.cli.state.project_sessions import canonical_cli_stores
+
+    _create_project_session("sess_global")
+    global_store = canonical_cli_stores()[-1]
+    global_store.add_message("sess.global", "user", "g-first")
+    global_store.add_message("sess.global", "assistant", "g-reply")
+
+    session_fork("sess.global", at_message=None, title=None)
+
+    from praisonaiagents.session.hierarchy import HierarchicalSessionStore
+
+    store = HierarchicalSessionStore(global_store.session_dir)
+    new_id = _forked_id(store, "sess.global")
+    assert new_id is not None
+    assert len(store._load_extended_session(new_id, force_reload=True).messages) == 2

src/praisonai-ts/SIGNATURE_PARITY.md

@@ -261,7 +261,7 @@ TS-only members: `options`?
 
 ### `Handoff.__init__`
 
-- Python: `src/praisonai-agents/praisonaiagents/agent/handoff.py:358`
+- Python: `src/praisonai-agents/praisonaiagents/agent/handoff.py:391`
 - TypeScript: `src/praisonai-ts/src/agent/handoff.ts:505` (ctor `src/praisonai-ts/src/agent/handoff.ts:759`)
 - Counts: 7 python params: 2 exact, 2 camelCase, 3 alias, 0 flattened, 0 missing; 2 mismatches; 2 waived; 13 TS-only of 21
 

src/praisonai-ts/signature-parity.json

@@ -3766,7 +3766,7 @@
         }
       ],
       "python": {
-        "location": "src/praisonai-agents/praisonaiagents/agent/handoff.py:358",
+        "location": "src/praisonai-agents/praisonaiagents/agent/handoff.py:391",
         "resolved_class": "Handoff"
       },
       "ts_only": [

src/praisonai/praisonai/app/agentos.py

@@ -115,11 +115,14 @@ class AgentOS:
             openapi_url=self.config.openapi_url,
         )
         
-        # Add CORS middleware
+        # Add CORS middleware. Credentials are never combined with a "*"
+        # origin: Starlette would reflect any caller's Origin back, letting
+        # arbitrary sites make credentialed requests.
+        cors_origins = list(self.config.cors_origins or [])
         app.add_middleware(
             CORSMiddleware,
-            allow_origins=self.config.cors_origins,
-            allow_credentials=True,
+            allow_origins=cors_origins,
+            allow_credentials="*" not in cors_origins,
             allow_methods=["*"],
             allow_headers=["*"],
         )

src/praisonai/praisonai/jobs/server.py

@@ -154,7 +154,8 @@ def create_app(
         app.add_middleware(
             CORSMiddleware,
             allow_origins=origins,
-            allow_credentials=True,
+            # Never pair credentials with "*" (Starlette reflects the Origin).
+            allow_credentials="*" not in origins,
             allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
             allow_headers=["Authorization", "Content-Type", "Origin", "Accept", "Idempotency-Key"],
         )

src/praisonai/tests/unit/app/test_agentos_cors.py

@@ -0,0 +1,30 @@
+"""AgentOS CORS must never pair a wildcard origin with credentials (#6094)."""
+
+import pytest
+
+pytest.importorskip("fastapi")
+from fastapi.testclient import TestClient  # noqa: E402
+
+from praisonai import AgentOS  # noqa: E402
+
+
+def _get(origins, origin, **headers):
+    app = AgentOS(name="t", agents=[], cors_origins=origins).get_app()
+    return TestClient(app).get("/health", headers={"Origin": origin, **headers})
+
+
+def test_wildcard_origin_disables_credentials():
+    # Older Starlette reflects the Origin whenever a cookie is present, so only
+    # the credentials header is version-independent here.
+    resp = _get(["*"], "https://evil.example", Cookie="s=1")
+    assert "access-control-allow-credentials" not in resp.headers
+
+    resp = _get(["*"], "https://evil.example")
+    assert resp.headers.get("access-control-allow-origin") == "*"
+    assert "access-control-allow-credentials" not in resp.headers
+
+
+def test_explicit_origin_keeps_credentials():
+    resp = _get(["https://ok.example"], "https://ok.example")
+    assert resp.headers.get("access-control-allow-origin") == "https://ok.example"
+    assert resp.headers.get("access-control-allow-credentials") == "true"