chronaxis.execution.temporal_buffer
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""以预先申请的预算批量推进独立时序检测器，并增量转交最终事件。23输入为ResourceBudget、冻结规则/会话域/固定来源引用和完整ConditionTick tuple；4可选StabilitySpec启用双向状态转换，迟滞可另传同刻退出条件，固定/批费用额外覆盖。5输出为须close的TemporalEventBatch。固定规则及完整refs元数据先计账再建立core。6每批按输入长度N申请ticks持有、NumPy数组/run掩码、最多N个Event及稳定ID构造7临时串的保守上界，随后调用一个独立core的真实批量push和正式drain_events。8返回批只保存事件tuple，不保存全量输入轴或core状态；整个本批预留转交结果，9直到close抵销，未用空间也保守保留，不声称精确RSS。旧批与后续推进相互独立。10活动批以各批对象内的双向指针登记，容器不建立哈希表，也不保留已关闭批的峰值槽位；11固定申请覆盖登记首指针，每批申请覆盖自己的链指针，稀疏关闭不留下未计账容量。12core开始前预算不足不改变检测状态，可重试；开始后任何异常或取消使owner失败，13丢弃内部core并禁止再次推进，保留此前已交付批，不声称状态事务回滚。14本组件不生成三视图/actual_unknown、不读来源、不调度deadline或发布公共AnalysisRunner；15source_refs是调用方给定的固定引用，不宣称覆盖输入每个CSV行的位置。16构造持续清理失败在原异常附temporal_detector_cleanup_owner，保留实际固定费用17和真实部分实例；不通过关闭调用方budget或上游查询弥补本次恢复责任。18"""1920# 输入及输出模型共享协议，两个检测器仍使用各自独立的状态/数组算法。21from ..core.models import SourceRef22from ..core.temporal import ConditionTick, TemporalRule23from ..core.temporal_reference import ReferenceTemporalDetector24from ..core.temporal_numpy import NumpyTemporalDetector25from ..core.stability import StabilitySpec, StabilityDetector26from ..core.start_end import StartEndSpec, StartEndDetector27from ..results.budget import ResourceBudget28from ..results.errors import AnalysisError, ResultClosedError29from ..results.spool import _cleanup_owner30# 泛型冻结元数据估算保留CSVSourceRef全部字段，不使用封闭结果codec缩掉定位。31from .history import _metadata_bytes323334class _BatchRegistry:35 """用已有批对象的双向指针登记活动结果，新增/移除不分配容器容量。3637 只保存首指针；每个成员的前后指针由其已申请的批对象持有。长度按需遍历，38 owner推进和关闭不计算长度。移除支持新增首节点尚未完全安装时的失败清理。39 """4041 def __init__(self):42 """建立固定大小登记器。参数: 无。返回: 新登记器。异常: MemoryError。"""43 # 空登记器只有首指针，不申请随历史批数增长的数组或哈希槽。44 self._first = None4546 def __bool__(self):47 """检查是否有活动批。参数: 无。返回: bool。异常: 无。"""48 # 生命周期循环只需首指针，避免关闭时遍历计数或复制集合。49 return self._first is not None5051 def __len__(self):52 """读取活动批数量。参数: 无。返回: int。异常: 内部链损坏。5354 该诊断接口按需遍历，不建立随批数增长的辅助计数容器。55 """56 # 在已有链上使用常量空间计数；正常推进不调用此诊断接口。57 count, result = 0, self._first58 while result is not None:59 count += 160 result = result._next61 # 计数不缓存，批关闭后没有高峰数量对应的遗留容量。62 return count6364 def __iter__(self):65 """遍历当前活动批。参数: 无。返回: 常量空间迭代器。异常: 内部链损坏。"""66 # 先保存下一节点，允许调用方在当前节点交付后关闭当前批。67 result = self._first68 while result is not None:69 following = result._next70 yield result71 result = following7273 def add(self, result):74 """把新批安装到首端，不分配额外节点。7576 参数: result尚未登记、具有_registry/_previous/_next指针的预算结果。77 用于时序事件批或内部预备条件结果，各自owner负责释放金额。78 返回: None。79 异常: ValueError为重复/跨登记器安装；取消由owner撤销部分安装。80 """81 # 每批只有一个登记归属，不把重复安装解释为创建第二份预算所有权。82 if result._registry is not None:83 raise ValueError('时序事件批已登记')84 # 先保留旧首端与归属，外层异常处理能识别并撤销任一部分安装步骤。85 result._previous, result._next, result._registry = None, self._first, self86 if self._first is not None:87 self._first._previous = result88 self._first = result8990 def discard(self, result):91 """移除一个完整或部分安装的批，保留其他活动批。9293 参数: result待移除批。返回: None；不属于本登记器时无副作用。94 异常: 内部链损坏；不抵销批预算，由其生命周期边界负责释放。95 """96 # 成员标记只用于本容器，避免失败清理影响先前已交付批的归属。97 if result._registry is not self:98 return99 # 每个邻接关系先确认实际指向本批，兼容新增首节点中途取消后的撤销。100 previous, following = result._previous, result._next101 if previous is not None and previous._next is result:102 previous._next = following103 if following is not None and following._previous is result:104 following._previous = previous105 if self._first is result:106 self._first = following107 # 脱链后不持有旧批或其邻接对象，关闭不会留下高峰容器容量。108 result._previous = result._next = result._registry = None109110111class TemporalEventBatch:112 """持有本批最终事件及独立申请，owner失败后仍可读至本批或owner关闭。113114 events为冻结tuple，事件模型不随后续core推进改变。外部另存已取得的事件115 超过close时刻须自行负责引用，不把释放申请解释为强制销毁外部对象。116 """117118 def __init__(self, owner, cost, events):119 """在完整预留中转交事件，不复制来源元数据或保留输入ticks。120121 参数: owner自有检测owner；cost本批已有申请；events已排空的Event tuple。122 返回: 新批结果，随后由owner登记。123 异常: MemoryError为构造失败，由owner清理本次申请并进入failed。124 """125 # 原事件及固定引用保持只读模型，预算归结果直到close。126 self._owner, self._cost, self._events, self._closed = owner, cost, events, False127 # 登记节点复用本批对象；字段随本批预留持有，不另外分配哈希表槽位。128 self._previous = self._next = self._registry = None129130 @property131 def closed(self):132 """读取批关闭状态。参数: 无。返回: bool。异常: 无。"""133 # 关闭标志即使事件引用已清理也可读取，支持幂等收尾。134 return self._closed135136 @property137 def events(self):138 """读取只读最终事件。参数: 无。返回: Event tuple。异常: ResultClosedError。"""139 # 关闭后的事件已不由本批预算持有，拒绝继续借用其私有引用。140 if self._closed:141 raise ResultClosedError('时序事件批已关闭')142 # 返回冻结tuple，后续core推进或其它批关闭不会改写本批事件。143 return self._events144145 def close(self):146 """抵销一次完整本批申请，不修改其它批或检测业务状态。147148 参数: 无。返回: None，重复关闭安全。149 异常: 内部账本不一致或取消；若已抵销则完成最小脱链/关闭后传播，150 尚未抵销时保持开放，可再次关闭。共享账本遵守其单线程契约。151 """152 # 第二次关闭不能再次抵销同一批申请。153 if self._closed:154 return155 # 单线程账本快照区分release调用入口取消和完成抵销后的返回取消。156 owner, cost = self._owner, self._cost157 budget, before = owner._budget, owner._budget.ram_bytes158 try:159 # 抵销和本地归零处于同一受保护块，后续脱链取消不会重复释放。160 budget.release_ram(cost)161 self._cost = 0162 owner._results.discard(self)163 self._closed = True164 self._events = self._owner = None165 self._previous = self._next = self._registry = None166 except BaseException:167 # 已抵销时完成最小收尾；未抵销时不删除仍有申请的登记批。168 if self._cost == 0 or budget.ram_bytes == before - cost:169 self._cost = 0170 owner._results.discard(self)171 self._closed = True172 self._events = self._owner = None173 self._previous = self._next = self._registry = None174 # 取消仍交还调用方，资源状态不会由吞异常伪装为正常调用。175 raise176177178class BudgetedTemporalDetector:179 """维护一个私有core、固定申请及活动事件批，预算外不累计旧events。180181 规则和domain只读，候选/run/前态跨普通批保存。每次push后正式drain输出，182 已交付事件由批独立计账；调用方不直接操作core。failed与closed分别表示183 推进失败和资源关闭，失败丢弃core但不隐式关闭先前批。184 """185186 def __init__(self, budget: ResourceBudget, rule: TemporalRule, start_ns: int, end_ns: int,187 source_refs: tuple[SourceRef, ...] = (), backend: str = 'numpy', *, stability=None, start_end=None, record_clock=False):188 """先申请完整规则/来源固定空间，再创建新的独立后端检测器。189190 参数: budget共享账本；rule冻结规则；start_ns/end_ns整数ns域；source_refs固定引用；191 backend为reference或numpy；stability或start_end分别选择防抖/迟滞或开始结束，互斥；192 record_clock=True时等待只由push的advances推进，默认仍按实际时间。193 返回: 新owner，未消费任何tick。194 异常: 类型/后端/规则/域/来源/预算/构造原错误；本次固定申请清理持续失败时，195 原异常附temporal_detector_cleanup_owner供恢复close，不改变调用方预算所有权。196 """197 # 固定申请之前只接受既有冻结协议，拒绝用任意对象驱动元数据遍历。198 if (not isinstance(budget, ResourceBudget) or not isinstance(rule, TemporalRule)199 or type(source_refs) is not tuple or any(not isinstance(ref, SourceRef) for ref in source_refs)):200 raise TypeError('预算时序检测需要ResourceBudget/TemporalRule及SourceRef tuple')201 # 后端选择显式且稳定；无静默回退或模型外数值实现。202 if type(backend) is not str or backend not in ('reference', 'numpy'):203 raise ValueError('时序后端必须为reference或numpy')204 if stability is not None and not isinstance(stability, StabilitySpec):205 raise TypeError('防抖参数需要StabilitySpec')206 if start_end is not None and (not isinstance(start_end, StartEndSpec) or stability is not None):207 raise TypeError('开始结束参数需要独立StartEndSpec，不能混用迟滞')208 # 估算只遍历调用方已提供的冻结模型，不建立事件集合、数组或core状态。209 metadata = _metadata_bytes((rule, source_refs))210 fixed = 32768 + 16 * _metadata_bytes((rule, start_ns, end_ns, source_refs, backend))211 if stability is not None:212 fixed += 32768 + 16 * _metadata_bytes(stability)213 if start_end is not None:214 fixed += 32768 + 16 * _metadata_bytes(start_end)215 # 先建立可幂等清理的空owner壳，尚不创建core或登记器。216 self._budget, self._fixed_cost, self._closed, self._failed = budget, 0, False, False217 self._rule = self._start = self._end = self._refs = self._detector = self._results = None218 self._event_metadata_cost = 0219 self._failed_workspace_cost = 0220 # 申请前快照覆盖reserve返回取消，避免已扣账但owner尚未记录申请的窗口。221 before = budget.ram_bytes222 try:223 # 固定申请成功后才允许建立登记器和两个独立后端中的一个。224 budget.reserve_ram(fixed)225 self._fixed_cost = fixed226 self._rule, self._start, self._end, self._refs = rule, start_ns, end_ns, source_refs227 self._event_metadata_cost = metadata228 self._results = _BatchRegistry()229 # 实际构造器继续检查域跨度/会话归属；参考不调用数组核，数组不回退参考。230 detector_type = ReferenceTemporalDetector if backend == 'reference' else NumpyTemporalDetector231 if start_end is not None:232 self._detector = StartEndDetector(rule, start_ns, end_ns, start_end, source_refs, backend=backend, record_clock=record_clock)233 elif stability is not None:234 self._detector = StabilityDetector(rule, start_ns, end_ns, stability, source_refs, backend=backend, record_clock=record_clock)235 else:236 self._detector = detector_type(rule, start_ns, end_ns, source_refs, record_clock=record_clock)237 except BaseException as exc:238 # reserve已经完成但返回时取消，仍将该申请纳入最小关闭收尾。239 if self._fixed_cost == 0 and budget.ram_bytes == before + fixed:240 self._fixed_cost = fixed241 # 部分构造通过同一关闭边界抵销；清理失败不能替换原错或丢失未交付owner。242 _cleanup_owner(self, exc, 'temporal_detector_cleanup_owner', '预算时序检测器构造')243 raise244245 def _ensure_open(self):246 """拒绝关闭后访问。参数: 无。返回: None。异常: ResultClosedError。"""247 # 资源已归还后不能继续使用私有规则或检测状态。248 if self._closed:249 raise ResultClosedError('预算时序检测器已关闭')250251 def _ensure_active(self):252 """拒绝关闭或失败后继续推进/读取deadline，不尝试恢复部分core状态。253254 参数: 无。返回: None。异常: ResultClosedError或AnalysisError为不可继续。255 """256 # 先处理资源生命周期，再区分不可重试的部分推进失败。257 self._ensure_open()258 if self._failed:259 raise AnalysisError('预算时序检测器已失败，禁止重试当前状态')260 if self._failed_workspace_cost:261 raise AnalysisError('上次工作区尚未释放，请先关闭预算时序检测器')262263 @property264 def rule(self):265 """读取冻结规则。参数: 无。返回: TemporalRule。异常: ResultClosedError。"""266 # 规则只借用固定申请内的冻结对象，不提供重新配置入口。267 self._ensure_open()268 return self._rule269270 @property271 def start_ns(self):272 """读取domain起点。参数: 无。返回: int ns。异常: ResultClosedError。"""273 # 域起点由构造器验证，读取时不重新计算或改变检测状态。274 self._ensure_open()275 return self._start276277 @property278 def end_ns(self):279 """读取domain终点。参数: 无。返回: int ns。异常: ResultClosedError。"""280 # 域终点由构造器验证，不能用读取请求补造terminal输入。281 self._ensure_open()282 return self._end283284 @property285 def next_deadline_ns(self):286 """读取core准确待确认截止。参数: 无。返回: int或None。异常: 关闭或失败。"""287 # 只从仍可推进的私有core读取准确deadline，不重建已失败状态。288 self._ensure_active()289 return self._detector.next_deadline_ns290291 @property292 def closed(self):293 """读取资源关闭状态。参数: 无。返回: bool。异常: 无。"""294 # 生命周期状态不依赖已释放的core或批引用。295 return self._closed296297 @property298 def failed(self):299 """读取推进失败状态。参数: 无。返回: bool。异常: 无；close不掩盖失败。"""300 # 保留失败事实供收尾诊断，资源关闭不能把它改写为成功。301 return self._failed302303 @property304 def duration_counts(self):305 """借出已结束迟滞核的固定累计值，其他检测器无额外结果。306307 参数: 无。返回: 整数纳秒对或None。异常: 关闭、失败或尚未结束错误。308 """309 self._ensure_active()310 return self._detector.duration_counts if isinstance(self._detector, StabilityDetector) else None311312 def extend_observed_end(self, end_ns):313 """将在线上界推进到新读到的真实末端，不交付terminal或分配输入历史。314315 参数: 已观测int64末端。返回: None。异常: 已封闭或末端回退。316 """317 from ..timebase.conversion import require_int64_ns318 self._ensure_active()319 end = require_int64_ns(end_ns, 'observed_end')320 if self._detector.ended or self._detector.finished or end < self._end:321 raise ValueError('在线观测上界只能在封闭前递增')322 self._detector.end_ns = self._end = end323 if isinstance(self._detector, StabilityDetector):324 self._detector._core.end_ns = end325326 def set_observed_end(self, end_ns):327 """将请求上界收窄为实际观测终点，必须在交付该终点组之前调用。328329 参数: end_ns实际int64终点。返回: None。不生成terminal或结果，不回撤已处理组。330 异常: 类型/范围或已完成错误；非法调用不改变core与owner边界。331 """332 from ..timebase.conversion import require_int64_ns333 self._ensure_active()334 end = require_int64_ns(end_ns,'observed_end')335 detector = self._detector336 if (detector.ended or detector.finished or not self._start <= end <= self._end337 or detector.last_time is not None and end <= detector.last_time):338 raise ValueError('观测终点必须在尚未交付的请求范围内')339 detector.end_ns = end340 if isinstance(detector,StabilityDetector):341 detector._core.end_ns = end342 self._end = end343344 def _workspace_cost(self, ticks, maximum_events):345 """估算输入持有、批量数组、最多N事件及其稳定ID临时串的同时存活上界。346347 参数: ticks已提供typed tuple；maximum_events本批最大输出数。348 返回: 非负int bytes，空批也覆盖固定工作容器。349 异常: 元数据估算错误；尚未调用core，不改变业务状态。350 """351 # 每tick最多一个最终事件；16KiB覆盖数组、前缀/run索引、掩码和Python装箱。352 # 规则/refs可能含任意长文本，32倍深成本同时覆盖Event模型/稳定ID编码及引用。353 # 输入tick深成本单独计入，已有调用方tuple不因此被当作免费内存。354 return 32768 + 4 * _metadata_bytes(ticks) + maximum_events * (16384 + 32 * self._event_metadata_cost)355356 def _register(self, cost, events):357 """只构造结果；外层取得本地引用后才登记并负责异常清理。358359 参数: cost本批完整申请；events已正式排空tuple。360 返回: 尚未登记的TemporalEventBatch。异常: 构造/返回取消由推进边界抵销并failed。361 """362 # helper返回取消时结果尚未安装；只有外层可见的local才能进入登记器。363 return TemporalEventBatch(self, cost, events)364365 def _discard_unreturned(self, result):366 """撤销本次未交付结果，预算仍由外层一次抵销，不关闭先前批。367368 参数: result为已构造的新批或None。返回: None。369 异常: 内部登记不一致；不调用result.close以免重复释放本批申请。370 """371 # 构造或helper返回失败时外层未取得对象，且对象还没有进入登记器。372 if result is None:373 return374 # add可能已经安装后报错，只撤销当前local；先前结果保持独立所有权。375 self._results.discard(result)376 result._closed, result._cost = True, 0377 result._events = result._owner = None378 result._previous = result._next = result._registry = None379380 def _fail(self):381 """丢弃部分推进的core并永久禁止重试，保留此前结果和固定元数据诊断。382383 参数: 无。返回: None。异常: 无主动异常；不声称旧业务状态已回滚。384 """385 # 业务状态不可事务回滚，失败后丢弃core而保留自有已交付批。386 self._failed = True387 self._detector = None388389 def _release_failed_workspace(self):390 """退还未交付批的实际申请，释放完成后取消也立即清槽。391392 参数: 无。返回: None。异常: 原退款错误；尚未释放的金额留给下一次close。393 本槽与固定申请、此前已交付批分开；只围绕一次直接账本调用判断副作用。394 """395 cost = self._failed_workspace_cost396 if not cost:397 return398 before = self._budget.ram_bytes399 try:400 self._budget.release_ram(cost)401 self._failed_workspace_cost = 0402 except BaseException:403 if self._budget.ram_bytes == before - cost:404 self._failed_workspace_cost = 0405 raise406407 def _refund_failed_workspace(self, cost, original):408 """接管本次未交付批的费用并尝试归还，保留原推进错误和最高owner。409410 参数: cost为本次已实际申请字节，original为业务或取消异常。411 返回: None。退款失败只追加说明；调用方继续抛原异常，close负责重试。412 """413 self._failed_workspace_cost = cost414 try:415 self._release_failed_workspace()416 except BaseException as cleanup:417 original.add_note('时序检测工作区释放失败: ' + type(cleanup).__name__)418 original.temporal_detector_cleanup_owner = self419420 def push(self, ticks: tuple[ConditionTick, ...], exit_ticks=None, *, advances=None) -> TemporalEventBatch:421 """在预算内整批推进独立core，随后正式排空当前最终事件。422423 参数: ticks完整组tuple，允许空批；迟滞可另给同刻exit_ticks，start_end必须提供结束组，普通防抖省略。424 批次不重置候选/边沿状态；advances仅记录模式提供，每组一个必要周期增量。425 返回: 须close的TemporalEventBatch，最多len(ticks)个最终事件。426 异常: 预算拒绝或申请取消发生在core前，可恢复；core调用或后续任何异常/取消使owner failed，427 抵销本批申请，先前批保持可读和计账。428 """429 # 输入协议检查在core前完成，错误输入不消耗任何检测状态。430 self._ensure_active()431 if type(ticks) is not tuple or any(not isinstance(tick, ConditionTick) for tick in ticks):432 raise TypeError('预算时序push需要完整ConditionTick tuple')433 if isinstance(self._detector, StartEndDetector) and exit_ticks is None:434 raise TypeError('start_end必须提供结束条件')435 if exit_ticks is not None and (not isinstance(self._detector, (StabilityDetector, StartEndDetector))436 or type(exit_ticks) is not tuple or any(not isinstance(tick, ConditionTick) for tick in exit_ticks)):437 raise TypeError('独立结束/退出仅供start_end或迟滞，且需完整ConditionTick tuple')438 cost = self._workspace_cost(ticks, len(ticks))439 from ..core.temporal_clock import require_clock_advances440 require_clock_advances(ticks, advances, self._detector.record_clock, self._detector.last_time)441 if advances is not None:442 cost += 4096 + len(advances) * 1024443 if exit_ticks is not None:444 cost += 4 * _metadata_bytes(exit_ticks)445 # 申请/开始标志与快照区分未推进的preflight取消和不能重试的core故障。446 before, reserved, started = self._budget.ram_bytes, False, False447 result = None448 try:449 # 申请失败不会触及core；已扣账的reserve返回取消也在外层一次抵销。450 self._budget.reserve_ram(cost)451 reserved = True452 # 从调用core开始即不承诺事务回滚；数组后端收到完整批而非逐点外部循环。453 started = True454 if exit_ticks is None:455 self._detector.push(ticks, advances=advances)456 else:457 self._detector.push(ticks, exit_ticks, advances=advances)458 events = self._detector.drain_events()459 # 单规则每tick最多闭合一个活动或生成一个点事件，超过即拒绝未计账输出。460 if len(events) > len(ticks):461 raise AnalysisError('时序检测输出超过预先申请的每tick一个事件上界')462 # 取得local后再安装，add中途失败也能精准撤销当前批的链接。463 result = self._register(cost, events)464 self._results.add(result)465 # 交付只读批，完整本批申请转交结果直到其close。466 return result467 except BaseException as error:468 # 排空/构造/登记属于推进后失败，preflight拒绝或取消则保留原core状态。469 if started:470 self._discard_unreturned(result)471 self._fail()472 # 正常申请记录与单线程快照共同覆盖reserve成功返回前的取消窗口。473 if reserved or self._budget.ram_bytes == before + cost:474 self._refund_failed_workspace(cost, error)475 raise476477 def finish(self) -> TemporalEventBatch:478 """在core已消费真正terminal后封闭检测器，返回剩余增量批且不重发旧事件。479480 参数: 无，不能补造终点或消费新的输入。481 返回: 须close的批；本owner每次push均排空，因此正常finish.events为空。482 异常: 预算拒绝/申请取消在core前可重试；缺terminal/重复finish或后续失败使owner failed。483 """484 # finish的固定工作容器也先申请，超限不消费terminal封闭状态。485 self._ensure_active()486 cost = self._workspace_cost((), 0)487 # 申请/开始标志覆盖reserve返回取消，外层local覆盖登记部分安装。488 before, reserved, started = self._budget.ram_bytes, False, False489 result = None490 try:491 # 预算申请失败或取消时不消费真正terminal的封闭状态。492 self._budget.reserve_ram(cost)493 reserved = True494 # 正式finish仍执行既有terminal协议；已交付事件不再留在core.events中。495 started = True496 self._detector.finish()497 events = self._detector.drain_events()498 # 每次push已经交付全部当前输出，finish不得重发事件或另造终点活动。499 if events:500 raise AnalysisError('预算检测器finish出现未在push中增量交付的事件')501 # 在外层可清理引用内完成安装后，才交付终点剩余批。502 result = self._register(cost, events)503 self._results.add(result)504 return result505 except BaseException as error:506 # terminal部分封闭不可重试，尚未开始的preflight仍保持原检测器可用。507 if started:508 self._discard_unreturned(result)509 self._fail()510 # 不对未成功的申请做释放，也不漏掉已扣账但返回取消的申请。511 if reserved or self._budget.ram_bytes == before + cost:512 self._refund_failed_workspace(cost, error)513 raise514515 def close(self):516 """关闭全部自有批和固定申请，不关闭调用方budget或删除任何数据。517518 参数: 无。返回: None，幂等且支持部分构造失败。519 异常: 内部账本不一致或取消；已释放固定申请时完成最小收尾后传播。520 failed标志保留用于说明此前失败；尚未释放时可重试关闭。521 """522 # 关闭可重复调用，不重复释放固定或批申请。523 if self._closed:524 return525 # 未交付批的失败退款不属于任何已交付结果，先处理它且保留重试金额。526 self._release_failed_workspace()527 # 逐项关闭避免收尾时再复制活动批集合；没有其它owner的结果进入此集合。528 while self._results:529 next(iter(self._results)).close()530 # 所有链节点都已脱离，仅剩固定申请；快照识别release完成后的返回取消。531 cost, before = self._fixed_cost, self._budget.ram_bytes532 try:533 self._budget.release_ram(cost)534 self._fixed_cost = 0535 self._closed = True536 self._rule = self._refs = self._detector = self._results = None537 except BaseException:538 # 固定抵销已发生则完成关闭，避免调用方重试时再释放同一申请。539 if self._fixed_cost == 0 or self._budget.ram_bytes == before - cost:540 self._fixed_cost = 0541 self._closed = True542 self._rule = self._refs = self._detector = self._results = None543 # 仍传播原异常，不把收尾完成当作此次close未发生取消。544 raise545