Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,26 @@ a started write without a provider locator stays unconfirmed after restart.
Revocation or changed identity prevents further delivery. Disable the existing
connection to revoke it; rollback must preserve its private delivery journals.

Newly verified App user mirrors also reuse the existing reaction journals:
`Get` acknowledges receipt; only an observed starting/running Core Turn adds
`OnIt`. Terminal completion, failure or interruption removes processing feedback
while retaining receipt feedback. A queued input never claims to be processing,
and assistant mirrors receive neither reaction. Long inputs use their last
verified concrete post, not the composite parts locator. Turn-only changes are
discovered within the existing five-second scan even without a transcript append.
Before each effect, the worker rechecks the canonical input/Turn, current source
and grants, and frozen provider message. Feedback failures retain recoverable
receipts without blocking message delivery or starting work. Unknown provider
writes are not retried. Optional App feedback runs on at most one slot of the
existing reply pool, outside the body outbox lock. Its final write merges only
feedback fields, preserving concurrently delivered current and later answers.
No synthetic incoming event or original-request
settlement is used. Existing finished mirrors are not backfilled; current active
mirrors can adopt the feedback path. `loopx chat --no-private-reactions` keeps
feedback disabled without changing the message path. The isolated HTTP/Core and
synthetic-provider checks qualify source behavior; installed Lark reactions and
actual provider presentation remain separate acceptance work.

Use **Continue original conversation** or the explicitly selected App steward
entry above to reach this path. Old ordinary history is retained separately.
This checkpoint covers text and bounded App image projection, not reply/thread
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -201,6 +201,19 @@ Core 重查 Session、当前绑定与受众,provider 核验原来源。消息
没有 provider locator 的写入,重启后仍保留未确认。撤权或身份变化阻止后续投递。
关闭既有连接即可撤销授权;回滚须保留私有投递 journal。

新核验送达的 App 用户镜像也复用已有表情回执:`Get` 表示已收到,仅在观测到 Core
Turn 已进入 starting/running 后添加 `OnIt`;完成、失败或中断后清理处理中表情,
保留已收到。排队不冒充处理中,App 答案不加这两种表情。长输入作用于最后一条已核验的
实际分段消息,不对组合 parts locator 操作。Turn 独立变化即使没有追加正文,也会由
既有五秒扫描发现。每次副作用前重查 canonical 输入/Turn、当前来源和授权,以及冻结
provider 消息;反馈失败保留可恢复回执,不阻断正文送达或启动工作,未知 provider
写入不重发。App 可选反馈最多占用既有回复池的一个槽位,在正文 outbox 锁外执行;
回写时仅合并反馈字段,保留并发完成的当前和后续答案投递。
不伪造 incoming event,也不结算原 Lark 请求。已有已结束镜像不补写表情,
当前活动镜像可接续此反馈。`loopx chat --no-private-reactions` 仍可关闭反馈,正文路径
不变。隔离真实 HTTP/Core 配合合成 provider 的检查只核验源码行为;安装后的实际
Lark 表情和 provider 呈现仍独立验收。

通过上述 **继续原会话** 或明确选定的 App 管家入口使用该路径;旧普通历史继续
独立保留。本检查点覆盖文本及有界 App 图片投影,不覆盖回复/话题关系镜像、向
Lark 历史回插旧时间或全渠道广播。Core 中的 App 图片复用既有资源上传与 post
Expand Down
26 changes: 21 additions & 5 deletions loopx/extensions/lark/goal_topic_runtime_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,8 @@ def _reconcile_private(self) -> None:
attention_due = 0.0
reports: Future[int] | None = None
reports_due = 0.0
feedback: Future[int] | None = None
feedback_due = 0.0
ingress: Future[list[str]] | None = None
ingress_due = time.monotonic() + 60
wakeup_scope = getattr(self.runtime_controller, "turn_release_wakeup", None)
Expand All @@ -101,10 +103,17 @@ def _reconcile_private(self) -> None:
with ThreadPoolExecutor(max_workers=4, thread_name_prefix="loopx-lark-private-reply") as pool, \
(wakeup_scope(self._delivery_wakeup) if callable(wakeup_scope) else nullcontext()):
def step() -> None:
nonlocal attempted, attention, attention_due, reports, reports_due, ingress, ingress_due
nonlocal attempted, attention, attention_due, reports, reports_due, ingress, ingress_due, feedback, feedback_due
# Clear BEFORE scanning: an arrival during the scan must also
# wake the following wait, rather than disappear in that gap.
self._delivery_wakeup.clear()
if feedback is not None and feedback.done():
try:
feedback.result()
except Exception:
logging.getLogger(__name__).warning("App input reaction reconciliation is pending")
feedback = None
feedback_due = time.monotonic() + 5
if reports is not None and reports.done():
try:
reports.result()
Expand Down Expand Up @@ -160,19 +169,26 @@ def step() -> None:
if report_provider is not None and reports is None and time.monotonic() >= reports_due:
reports_due = time.monotonic() + 30
for path in sorted(paths, key=lambda path: attempted.get(path, 0)):
if (len(pending) + int(reports is not None or reports_ready) + int(attention is not None or attention_ready)
if (len(pending) + int(feedback is not None) + int(reports is not None or reports_ready) + int(attention is not None or attention_ready)
+ int(ingress is not None or ingress_ready) >= 4 or self._closed.is_set()):
break
if path in pending:
continue
pending[path] = pool.submit(transport.reconcile_request, path)
attempted[path] = attempted.get(path, 0) + 1
if reports_ready and len(pending) + int(attention is not None or attention_ready) + int(ingress is not None or ingress_ready) < 4 and not self._closed.is_set():
if reports_ready and len(pending) + int(feedback is not None) + int(attention is not None or attention_ready) + int(ingress is not None or ingress_ready) < 4 and not self._closed.is_set():
reports = pool.submit(report_provider.reconcile)
if attention_ready and len(pending) + int(reports is not None) + int(ingress is not None or ingress_ready) < 4 and not self._closed.is_set():
if attention_ready and len(pending) + int(feedback is not None) + int(reports is not None) + int(ingress is not None or ingress_ready) < 4 and not self._closed.is_set():
attention = pool.submit(owner_attention.reconcile)
if ingress_ready and len(pending) + int(reports is not None) + int(attention is not None) < 4 and not self._closed.is_set():
if ingress_ready and len(pending) + int(feedback is not None) + int(reports is not None) + int(attention is not None) < 4 and not self._closed.is_set():
ingress = pool.submit(observer.observe, cancelled=self._closed.is_set)
app_feedback = getattr(transport, "reconcile_app_feedback", None)
# Optional feedback uses one slot in this same bounded
# pool and leaves capacity for canonical body deliveries.
if (callable(app_feedback) and feedback is None and time.monotonic() >= feedback_due
and len(pending) + int(reports is not None) + int(attention is not None)
+ int(ingress is not None) < 3 and not self._closed.is_set()):
feedback = pool.submit(app_feedback)
except Exception:
logging.getLogger(__name__).warning("Private Chat delivery discovery is pending")

Expand Down
82 changes: 73 additions & 9 deletions loopx/extensions/lark/inbox_reactions.py
Original file line number Diff line number Diff line change
Expand Up @@ -626,6 +626,7 @@ def ensure_lark_event_inbox_received_reaction_locked(
def _ensure_reaction_creation_locked(
*, inbox: Path, message_id: str, emoji_type: str, reaction_phase: str,
create_reaction: ReactionCreator, delete_reaction: ReactionDeleter,
captured_pending: bool = True,
) -> dict[str, Any]:
"""One provider creation lifecycle for received and processing feedback.

Expand All @@ -642,23 +643,23 @@ def _ensure_reaction_creation_locked(
status=f"already_{reaction_phase}",
ok=True,
configured=True,
captured_pending=True,
captured_pending=captured_pending,
)
operation = _load_reaction_creation_operation(inbox=inbox, message_id=message_id, reaction_phase=reaction_phase)
if operation and operation["emoji_type"] != emoji_type:
return _received_reaction_result(
status="operation_config_changed",
ok=False,
configured=True,
captured_pending=True,
captured_pending=captured_pending,
blocker=f"lark_inbox_{reaction_phase}_reaction_operation_config_changed",
)
if operation and operation["phase"] == "prepared":
return _received_reaction_result(
status="provider_outcome_uncertain",
ok=False,
configured=True,
captured_pending=True,
captured_pending=captured_pending,
blocker=f"lark_inbox_{reaction_phase}_reaction_provider_outcome_uncertain",
)
if operation and operation["phase"] == "created":
Expand All @@ -676,7 +677,7 @@ def _ensure_reaction_creation_locked(
status="receipt_failed",
ok=False,
configured=True,
captured_pending=True,
captured_pending=captured_pending,
external_writes_performed=False,
blocker=f"lark_inbox_{reaction_phase}_reaction_receipt_failed",
)
Expand All @@ -685,7 +686,7 @@ def _ensure_reaction_creation_locked(
status="receipt_recovered",
ok=True,
configured=True,
captured_pending=True,
captured_pending=captured_pending,
)
_write_reaction_creation_operation(
inbox=inbox,
Expand All @@ -700,7 +701,7 @@ def _ensure_reaction_creation_locked(
status="failed",
ok=False,
configured=True,
captured_pending=True,
captured_pending=captured_pending,
blocker=f"lark_inbox_{reaction_phase}_reaction_create_failed",
)
try:
Expand All @@ -720,7 +721,7 @@ def _ensure_reaction_creation_locked(
status="operation_receipt_failed",
ok=False,
configured=True,
captured_pending=True,
captured_pending=captured_pending,
created_count=int(not cleaned_up),
external_writes_performed=True,
blocker=(
Expand All @@ -742,7 +743,7 @@ def _ensure_reaction_creation_locked(
status="receipt_failed",
ok=False,
configured=True,
captured_pending=True,
captured_pending=captured_pending,
created_count=1,
external_writes_performed=True,
blocker=f"lark_inbox_{reaction_phase}_reaction_receipt_failed",
Expand All @@ -752,11 +753,74 @@ def _ensure_reaction_creation_locked(
status=reaction_phase,
ok=True,
configured=True,
captured_pending=True,
captured_pending=captured_pending,
created_count=1,
)


def _reconcile_outbound_input_reactions(
*, project: str | Path, config_path: str | Path, message_id: str,
processing: bool, complete: bool, before_write: Callable[[], bool],
runner: CommandRunner,
) -> dict[str, Any]:
"""Reuse reaction receipts for an independently verified outbound input.

This is not an Inbox capture or settlement. The private-conversation
caller supplies fresh canonical/source and provider-message verification.
"""
config = load_lark_event_inbox_config(project=project, config_path=config_path)
if not config["enabled"] or not MESSAGE_ID_PATTERN.fullmatch(message_id):
raise ValueError("outbound feedback requires an enabled inbox and exact message")
if config["reply"].get("received_reaction_policy") != "retain":
raise ValueError("outbound feedback requires retained receipt feedback")
inbox = config["inbox_path"]
with lark_inbox_reaction_lock(inbox=inbox, message_id=message_id):
if not before_write():
raise ValueError("outbound feedback authority unavailable")
created_count = 0
writes_performed = False
for phase in ("received", "processing") if processing and not complete else ("received",):
emoji = str(config["reply"].get(f"{phase}_reaction_emoji") or "")
if not emoji:
continue

def create(target: str, value: str) -> str | None:
if not before_write():
# This callback is reached only for a newly prepared
# creation. No provider call occurred; old unknown writes
# never reach it and keep their original journal.
_clear_reaction_creation_operation(inbox=inbox, message_id=target, reaction_phase=phase)
return None
return _create_reaction(runner=runner, profile=config["reply"]["sender_profile"],
message_id=target, emoji_type=value)

result = _ensure_reaction_creation_locked(inbox=inbox, message_id=message_id,
emoji_type=emoji, reaction_phase=phase, captured_pending=False,
create_reaction=create,
delete_reaction=lambda target, reaction: before_write() and _delete_reaction(
runner=runner, profile=config["reply"]["sender_profile"], message_id=target, reaction_id=reaction))
created_count += result["created_count"]
writes_performed |= result["external_writes_performed"]
if not result["ok"]:
result["created_count"] = created_count
result["external_writes_performed"] = writes_performed
return result
if complete:
def authorized_runner(args: Sequence[str]) -> Mapping[str, Any]:
return runner(args) if before_write() else {"returncode": 1}

result = _complete_lark_event_inbox_reactions_locked(config=config,
message_id=message_id, execute=True, runner=authorized_runner)
else:
result = _operation_result(operation="processing" if processing else "received",
status="processing" if processing else "received", ok=True, execute=True,
configured=bool(config["reply"].get("received_reaction_emoji")
or config["reply"].get("processing_reaction_emoji")))
result["created_count"] += created_count
result["external_writes_performed"] |= writes_performed
return result


def ensure_lark_event_inbox_received_reaction(
*,
project: str | Path,
Expand Down
Loading
Loading