Quick Answer: รวมเฉพาะคำขอที่ยังอยู่ระหว่างทำงาน
Single-flight คือการรวมคำขออ่านที่มีความหมายและสิทธิ์เทียบเท่ากัน ขณะงานแรกยัง in-flight (เริ่มแล้วแต่ยังไม่จบ): คำขอแรกเปิด flight และคำขอซ้ำเข้าร่วมรอผล. เมื่อจบด้วย OK หรือ ERROR ระบบส่งผลให้ผู้รอแล้วลบ flight.
กลไกนี้ไม่ใช่ cache หลังงานเสร็จ, retry, queue, rate limit หรือการประสานหลาย process. บทเรียนจำลอง event แบบลำดับด้วย Python; ไม่เรียก Tool, API หรือ LLM จริง. Go x/sync v0.24.0 docs ระบุการรวมคำเรียกซ้ำระหว่างทำงานและเตือนว่าไม่ใช่รุ่นล่าสุด.
ในวงจรจาก LLM chat สู่ Tool calling โมเดลอาจเสนอ structured call แต่ orchestrator ต้องตรวจ schema และสิทธิ์ก่อนเรียกบริการ. จุดรวมคำขออยู่ในชั้นแอปหลังตรวจ caller; ไม่ควรให้โมเดลเลือกว่าผู้ใช้ใดแชร์ผลกันได้.
ผลการเรียนรู้ที่สังเกตได้
หลังอ่านจบ ผู้อ่านจะจำแนก START ว่าเป็นคำขอแรกที่เปิด flight, JOIN ว่าเป็นผู้เรียกเพิ่มที่ใช้ key เดียวกันขณะ flight ยัง pending, settle ว่าเป็นการปิด flight ด้วยผล OK หรือ ERROR และ delivery ว่าเป็นการส่งผลให้ waiters ที่ลงทะเบียนไว้แล้วได้. คำว่า pending ในที่นี้หมายถึงงานยังไม่ถูกปิดใน state ของแบบจำลอง.
ผู้อ่านจะเขียน key สำหรับงานอ่านที่มี tool, trusted scope, normalized query และ schema revision ครบทั้งสี่มิติ พร้อมอธิบายว่าการละมิติหนึ่งอาจรวมคำขอที่มีสิทธิ์หรือผลลัพธ์ต่างกันอย่างไร. การออกแบบ key เป็นการแบ่งกลุ่มคำขอ ไม่ใช่การพิสูจน์ตัวตนหรืออนุมัติสิทธิ์.
สิ่งส่งมอบที่ตรวจด้วยมือคือ JSON fixture, hand trace, source code และ acceptance checklist. ผู้ตรวจสามารถเทียบ flight id กับผู้รับหลังทุก event และเห็นว่า F3 ใหม่เกิดหลัง cleanup; สิ่งนี้ตรวจ contract ที่เขียนไว้ ไม่ได้ยืนยัน runtime behavior.
จาก LLM chat สู่คำขอ Tool ที่ซ้ำกัน
ใน LLM chat ผู้ใช้มักส่งข้อความหนึ่งครั้งแล้วอ่านคำตอบหนึ่งชุด แต่ Agent ที่เชื่อม Tool อาจแยกงานเป็นหลายขั้น: โมเดลเลือกค้นแคตตาล็อก, แอปตรวจ argument, Tool อ่าน API, แล้วผลกลับเข้าบริบทให้โมเดลตอบ หากผู้ใช้สองคนหรือสอง branch ของ workflow ขอข้อมูล SKU เดียวกันในช่วงที่การอ่านครั้งแรกยังช้า ระบบอาจส่ง read request ซ้ำโดยไม่จำเป็น.
ตัวอย่างเช่น ผู้ช่วยฝ่ายขายสองหน้าจออ่านราคา SKU-7 จากแคตตาล็อกเดียวกันในเวลาใกล้กัน การเรียกแรกยังรอ network อยู่ คำขอที่สองมาถึงก่อน response กลับ หากนโยบายสิทธิ์และรูปแบบคำตอบเหมือนกัน แอปอาจให้คำขอที่สอง join flight แรกและใช้ response เดียวกันได้. จุดที่ต้องพิสูจน์คือคำขอทั้งคู่ขอข้อมูลเดียวกันจริงและผู้รับทั้งคู่ได้รับอนุญาตให้เห็นผลเดียวกัน.
Cache เก็บผลหลังการอ่านจบและต้องมีอายุ, invalidation และการตรวจสิทธิ์ต่อ hit; single-flight เก็บเพียงการอ่านที่ยังไม่จบ. Bulkhead แบ่งขอบเขตงานค้างหรือคิว ส่วน retry ตัดสินใจลองใหม่หลัง failure. อย่าใช้คำเหล่านี้แทนกันเพราะแต่ละกลไกแก้คนละจังหวะ.
กลไก: START, JOIN, settle และ read key
Read key คือ tuple (tool, trusted_scope, normalized_query, schema_version). สี่มิตินี้แยกชนิด Tool, ขอบเขตสิทธิ์, query ที่ normalize ตาม contract และรูปแบบผล. schema_version ไม่ใช่ data snapshot; fixture ไม่มี snapshot pin หากงานจริงต้องอ่าน snapshot เดียวกัน ให้เพิ่ม trusted snapshot ID ใน key และ Tool contract.
START เปิด flight เมื่อ key ยังว่างและลง waiter แรก; JOIN เพิ่ม waiter เมื่อ key เดียวกันยัง pending. flight คือรายการงานที่เปิดรอผล; pending คือยังไม่ settle. W1–W8 เป็น label ของคำขอที่รอ และหนึ่ง label ห้ามแทนสองคำขอที่ pending พร้อมกัน. จำกัดไม่เกินสี่ waiters ต่อ flight และสอง flight รวม.
settle ตรวจ alias, flight_id ที่เป็น int จริงและตรงกับ flight ปัจจุบัน, outcome OK/ERROR และ payload ที่พิมพ์ได้ 1–80 ตัวอักษรก่อนแก้ state. จากนั้นสร้าง frozen Result ที่มี waiters เป็น tuple ส่งผลเดียวกันและลบ entry. ERROR ก็ล้าง flight; caller เป็นผู้กำหนด retry แยก.
หลัง cleanup key กลับเป็น absent จึง START รหัสใหม่เมื่อมี request รอบถัดไป. stale หรือ unknown flight_id ปฏิเสธโดยไม่ลบ flight ใหม่. รหัสเพิ่มใน process นี้ไม่ใช่ fencing token, exactly-once หรือ coordination ข้าม worker.
ชั้น identity ต้อง authorize caller ก่อนค้น active map ทุกครั้ง. นอกจาก key ต้องตรวจว่า waiter label ยังไม่อยู่ใน flight ใดที่ pending; การปฏิเสธเกิดก่อน mutation. A/B เป็น trusted aliases คงที่และไม่ใช่ identity system.
อ่าน lifecycle เป็น ABSENT → START → JOIN ซ้ำได้ระหว่าง pending → settle OK/ERROR → fan-out และ cleanup → ABSENT. Request หลัง cleanup เปิด START ใหม่; ERROR ไม่ทำให้ entry ค้าง.
| แนวคิด | ความหมายในบทนี้ | จุดตรวจ |
|---|---|---|
| START | เปิด flight ใหม่เมื่อ read key ยังไม่มีงานค้าง | ได้ flight_id ใหม่และ waiter แรก |
| JOIN | เพิ่ม waiter เมื่อ key เดียวกันยัง pending | ใช้ flight_id เดิม ไม่มี START เพิ่ม |
| settle | ปิดงานด้วย OK หรือ ERROR | ต้องตรงกับ flight_id ปัจจุบัน |
| cleanup | ลบ active entry หลังส่งผล | คำขอถัดไปได้รหัสใหม่ |
| read key | tool + trusted scope + normalized query + schema version | ครบทุกมิติที่เปลี่ยนสิทธิ์หรือผล |
| waiter | ตัวแทนผู้เรียกที่รอผลใน flight | ห้ามซ้ำใน flight และจำกัดจำนวน |
| flight_id | เลขท้องถิ่นเพิ่มต่อ START | ไม่ใช่รหัสประสานงานข้าม process |
| payload | string จำลองผลหรือ error สั้น | immutable ใน result; จำกัด 1–80 ตัวอักษร |
Worked example: ไล่แปดเหตุการณ์ด้วยมือ
A คือ (catalog.lookup, team-a/view-r7, sku-7, catalog.v1); B ต่างเฉพาะ trusted_scope เป็น team-b/view-r7. W1 เปิด A/F1; W2,W3 join; W4 เปิด B/F2 เพราะ scope ต่าง.
settle A/F1 ส่ง OK ให้ W1–W3 แล้วลบ A. W5 จึงเปิด A/F3 ใหม่ ไม่ได้อ่าน completed cache. B/F2 ส่ง ERROR ให้ W4 แล้วลบ B โดยไม่ retry; A/F3 ส่งผล `catalog price 101 (mock)` ให้ W5. ค่า 100 และ 101 เป็นข้อความจำลอง ไม่ได้ยืนยัน snapshot เดียวกัน.
ตารางมีห้า request intents: START 3, JOIN 2; settlements ส่งถึงผู้รอห้าราย. จำนวน Tool/API/LLM calls จริงเป็นศูนย์. ไม่มีการวัด latency, cost หรือ performance.
เมื่ออ่านตารางให้ตาม key และ flight id ไม่ใช่ตาม query อย่างเดียว. Event 5 กับ event 6 แสดงขอบเขตระหว่าง in-flight sharing กับ request หลังงานจบ; event 7 ยืนยันว่า ERROR ก็ terminal สำหรับ flight นั้น. event ไม่เรียก catalog จริง.
| # | event | ผล | active หลัง event | ผู้รับ/หมายเหตุ |
|---|---|---|---|---|
| 1 | request A / W1 | START F1 | A:F1 → W1 | เปิด flight A |
| 2 | request A / W2 | JOIN F1 | A:F1 → W1,W2 | ไม่มี START เพิ่ม |
| 3 | request A / W3 | JOIN F1 | A:F1 → W1,W2,W3 | ใช้ผล flight เดียวกัน |
| 4 | request B / W4 | START F2 | A:F1; B:F2 → W4 | scope ต่างจึงแยก |
| 5 | settle A/F1 OK | DELIVER | B:F2 → W4 | ส่ง W1,W2,W3; ลบ A |
| 6 | request A / W5 | START F3 | B:F2; A:F3 → W5 | ผลเดิมไม่ถูก cache |
| 7 | settle B/F2 ERROR | DELIVER | A:F3 → W5 | ส่ง error W4; ลบ B; ไม่ retry |
| 8 | settle A/F3 OK | DELIVER | ว่าง | ส่ง W5; ลบ A |
ขอบเขตสถาปัตยกรรมและแหล่งข้อมูล
Go x/sync/singleflight เป็นแหล่งอ้างอิงของพฤติกรรม duplicate function call ใน implementation ของ Go: การเรียกซ้ำระหว่างงานค้างใช้ผลของการทำงานเดียว และ API มีรูปแบบ Do, DoChan และ Forget ตามเอกสาร. เอกสารที่อ่านได้เป็น v0.24.0 และมีคำเตือนว่าไม่ใช่ latest module version. บทความนำมาเป็นหลักฐานว่ารูปแบบ single-flight มี implementation library ที่บันทึกไว้ ไม่ใช่หลักฐานว่า Python mock มี semantics เหมือนกันทุกประการ.
Python 3.12 dataclasses อธิบายว่า frozen dataclass ป้องกันการกำหนด field ผ่านกลไกปกติ แต่ไม่ได้เป็นระบบตรวจ annotation เป็น runtime type checker. โค้ดจึงตรวจ exact type ของ input อย่างชัดเจนและเก็บ waiters ของ Result เป็น tuple. ข้อกำหนดนี้จำกัดการแก้ไขโดยผู้เรียกทั่วไปในแบบฝึก แต่ไม่ได้อ้าง immutability แบบป้องกันทุกวิธีหรือเหมาะกับข้อมูลทุกชนิด.
Go docs เป็น DOCUMENTED BEHAVIOR ของเอกสาร v0.24.0 ที่ตรวจ ไม่ใช่คำยืนยันว่าเป็นรุ่นล่าสุด. การแยก key ตาม trusted scope เป็น ENGINEERING ANALYSIS; event และราคาใน fixture เป็น SIMULATION. Python 3.12 dataclass frozen จำกัดการกำหนด field ปกติ แต่ annotation ไม่ได้ทำ runtime type validation.
Implementation walkthrough: Python 3.12+ mock
`code/singleflight_mock.py` ใช้ Python 3.12+ standard library และ registry A/B คงที่. SingleFlight เก็บ active map กับ local ID; Result เป็น frozen dataclass และเก็บ recipients เป็น tuple. Frozen จำกัดการกำหนด field ปกติ แต่ annotation ไม่ได้ตรวจ type ตอน runtime.
CLI รับ JSON `{events:[…]}` 1–32 events; request และ settle มี exact fields ตาม README. Registry ไม่รับ scope จาก input. ตัวตรวจปฏิเสธ duplicate JSON fields, alias/operation ที่ไม่รู้จัก, waiter ที่กำลัง pending, stale ID และ payload ผิดก่อน mutation.
Mock บันทึก START/JOIN/DELIVER และ state เพื่ออ่าน trace; ไม่มี file write, network, secret, LLM หรือ Tool invocation. Return value คัดลอก waiter list เพื่อไม่ให้ผู้เรียกแก้ event history ได้. ใช้ไฟล์เต็มประกอบ walkthrough; การรันยัง NOT RUN.
Parser ปฏิเสธ duplicate JSON keys และ object ที่มี field ขาดหรือเกิน; operation รับเฉพาะ request หรือ settle. `flight_id` ต้องเป็น positive int จริง ไม่รับ bool ที่สืบทอดจาก int. Payload ที่รับเป็น string สั้นและพิมพ์ได้; ไม่มี object mutable ที่แชร์ข้ามผู้รอ.
deliveries เป็นบันทึกผลของการสาธิต จึงยังปรากฏใน snapshot แต่ request ค้นเฉพาะ active entry และไม่หยิบบันทึกนี้มาตอบแทนการเริ่มเที่ยวใหม่ บันทึกประวัติจึงไม่ได้ทำหน้าที่เป็น result cache
singleflight_mock.py — Mock / NOT RUN
"""Deterministic teaching model of in-process single-flight read coalescing.
Records START/JOIN/settle intents. Does not call a tool or schedule concurrent work.
"""
from __future__ import annotations
from dataclasses import dataclass
import json
import sys
from typing import Any
class ContractError(ValueError):
"""Input violates the bounded mock contract."""
@dataclass(frozen=True)
class Result:
flight_id: int
outcome: str
payload: str
waiters: tuple[str, ...]
REGISTRY = {
"A": ("catalog.lookup", "team-a/view-r7", "sku-7", "catalog.v1"),
"B": ("catalog.lookup", "team-b/view-r7", "sku-7", "catalog.v1"),
}
VALID_WAITERS = {f"W{i}" for i in range(1, 9)}
MAX_WAITERS = 4
MAX_FLIGHTS = 2
def _exact_keys(value: Any, keys: set[str], label: str) -> None:
if type(value) is not dict or set(value) != keys:
raise ContractError(f"{label}: expected exactly {sorted(keys)}")
def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise ContractError(f"duplicate JSON field: {key}")
result[key] = value
return result
def _payload(value: Any) -> None:
if type(value) is not str or not 1 <= len(value) <= 80:
raise ContractError("payload must be a string of 1..80 printable characters")
if any(not ch.isprintable() for ch in value):
raise ContractError("payload must be printable and contain no controls")
class SingleFlight:
"""One active record per fixed, trusted read key in this process."""
def __init__(self) -> None:
self._active: dict[tuple[str, str, str, str], dict[str, Any]] = {}
self._next_id = 1
self.deliveries: list[Result] = []
self.events: list[dict[str, Any]] = []
def request(self, alias: Any, waiter: Any) -> dict[str, Any]:
if type(alias) is not str or alias not in REGISTRY:
raise ContractError("alias must be a trusted registry name A or B")
if type(waiter) is not str or waiter not in VALID_WAITERS:
raise ContractError("waiter must be a registered label W1..W8")
if any(waiter in item["waiters"] for item in self._active.values()):
raise ContractError("waiter already has a pending request")
key = REGISTRY[alias]
current = self._active.get(key)
if current is not None:
if len(current["waiters"]) >= MAX_WAITERS:
raise ContractError("flight waiter capacity reached")
current["waiters"].append(waiter)
event = {"event": "JOIN", "alias": alias, "flight_id": current["id"], "waiter": waiter}
else:
if len(self._active) >= MAX_FLIGHTS:
raise ContractError("active flight capacity reached")
flight_id = self._next_id
self._next_id += 1
current = {"id": flight_id, "waiters": [waiter]}
self._active[key] = current
event = {"event": "START", "alias": alias, "flight_id": flight_id, "waiter": waiter}
self.events.append(event)
return dict(event)
def settle(self, alias: Any, flight_id: Any, outcome: Any, payload: Any) -> dict[str, Any]:
if type(alias) is not str or alias not in REGISTRY:
raise ContractError("alias must be a trusted registry name A or B")
if type(flight_id) is not int or flight_id < 1:
raise ContractError("flight_id must be an exact positive integer")
if type(outcome) is not str or outcome not in {"OK", "ERROR"}:
raise ContractError("outcome must be OK or ERROR")
_payload(payload)
key = REGISTRY[alias]
current = self._active.get(key)
if current is None or current["id"] != flight_id:
raise ContractError("unknown or stale flight_id")
result = Result(flight_id, outcome, payload, tuple(current["waiters"]))
event = {"event": "DELIVER", "alias": alias, "flight_id": flight_id,
"outcome": outcome, "payload": payload, "waiters": list(result.waiters)}
self.deliveries.append(result)
del self._active[key]
self.events.append(event)
return {**event, "waiters": list(result.waiters)}
def snapshot(self) -> dict[str, Any]:
active = []
for key, value in self._active.items():
alias = "A" if key == REGISTRY["A"] else "B"
active.append({"alias": alias, "flight_id": value["id"], "waiters": list(value["waiters"])})
return {"active": active, "next_flight_id": self._next_id,
"deliveries": [{"flight_id": r.flight_id, "outcome": r.outcome,
"payload": r.payload, "waiters": list(r.waiters)} for r in self.deliveries]}
def _run(events: Any) -> dict[str, Any]:
if type(events) is not list or not 1 <= len(events) <= 32:
raise ContractError("events must be a list of 1..32 operations")
model = SingleFlight()
output = []
for index, event in enumerate(events, 1):
if type(event) is not dict or type(event.get("op")) is not str:
raise ContractError(f"event {index}: operation object required")
if event["op"] == "request":
_exact_keys(event, {"op", "alias", "waiter"}, f"event {index}")
output.append(model.request(event["alias"], event["waiter"]))
elif event["op"] == "settle":
_exact_keys(event, {"op", "alias", "flight_id", "outcome", "payload"}, f"event {index}")
output.append(model.settle(event["alias"], event["flight_id"], event["outcome"], event["payload"]))
else:
raise ContractError(f"event {index}: op must be request or settle")
return {"events": output, "state": model.snapshot(), "actual_tool_api_llm_calls": 0}
def main() -> int:
try:
document = json.loads(sys.stdin.read(), object_pairs_hook=_unique_object)
_exact_keys(document, {"events"}, "input")
response = _run(document["events"])
except (json.JSONDecodeError, ContractError) as exc:
print(json.dumps({"error": str(exc)}, ensure_ascii=False))
return 2
print(json.dumps(response, ensure_ascii=False, separators=(",", ":")))
return 0
if __name__ == "__main__":
raise SystemExit(main())
ตรวจรับจากสัญญาและไฟล์ทดสอบที่เขียนไว้
ตรวจแปด event ตามตาราง: START A/F1 ให้ W1; JOIN W2,W3; START B/F2 ให้ W4; settle A/F1 OK ส่ง W1–W3 และลบ A; request A/W5 เปิด F3; settle B/F2 ERROR ส่ง W4 และลบ B; settle A/F3 OK ส่ง W5 และลบ A. สุดท้าย active ว่างและ next ID เป็น 4.
test ครอบคลุม lifecycle, START ใหม่หลัง ERROR, stale ID, scope separation, alias ที่ไม่น่าเชื่อถือ, waiter ซ้ำข้าม flight, เพดานสี่ waiters, duplicate JSON fields, exact CLI schema และการไม่ alias delivery list. Registry มีเพียง A/B จึงไม่ทดสอบการเกินเพดานสอง active flights.
Testing status: NOT RUN; ไม่ได้ import, execute, compile หรือรัน unittest. `expected-output.json` เป็น hand trace. Lab UI ยัง UNKNOWN; mock ไม่พิสูจน์ Sandbox, Live, concurrent race, provider behavior, latency หรือ cost.
Acceptance เพิ่มเติม: invalid alias, waiter ที่กำลังรอ, duplicate JSON field, payload ที่มี control character และ stale settle ต้องไม่เปลี่ยน active map. เมื่อ delivery คืนรายการ waiters การแก้ list ที่คืนให้ caller ต้องไม่แก้ event history หรือ tuple ใน Result.
Input จำลอง
{"events":[{"op":"request","alias":"A","waiter":"W1"},{"op":"request","alias":"A","waiter":"W2"},{"op":"request","alias":"A","waiter":"W3"},{"op":"request","alias":"B","waiter":"W4"},{"op":"settle","alias":"A","flight_id":1,"outcome":"OK","payload":"catalog price 100 (mock)"},{"op":"request","alias":"A","waiter":"W5"},{"op":"settle","alias":"B","flight_id":2,"outcome":"ERROR","payload":"mock_unavailable"},{"op":"settle","alias":"A","flight_id":3,"outcome":"OK","payload":"catalog price 101 (mock)"}]}
Expected output — คำนวณด้วยมือ ไม่ใช่ผลรัน
{"events":[{"event":"START","alias":"A","flight_id":1,"waiter":"W1"},{"event":"JOIN","alias":"A","flight_id":1,"waiter":"W2"},{"event":"JOIN","alias":"A","flight_id":1,"waiter":"W3"},{"event":"START","alias":"B","flight_id":2,"waiter":"W4"},{"event":"DELIVER","alias":"A","flight_id":1,"outcome":"OK","payload":"catalog price 100 (mock)","waiters":["W1","W2","W3"]},{"event":"START","alias":"A","flight_id":3,"waiter":"W5"},{"event":"DELIVER","alias":"B","flight_id":2,"outcome":"ERROR","payload":"mock_unavailable","waiters":["W4"]},{"event":"DELIVER","alias":"A","flight_id":3,"outcome":"OK","payload":"catalog price 101 (mock)","waiters":["W5"]}],"state":{"active":[],"next_flight_id":4,"deliveries":[{"flight_id":1,"outcome":"OK","payload":"catalog price 100 (mock)","waiters":["W1","W2","W3"]},{"flight_id":2,"outcome":"ERROR","payload":"mock_unavailable","waiters":["W4"]},{"flight_id":3,"outcome":"OK","payload":"catalog price 101 (mock)","waiters":["W5"]}]},"actual_tool_api_llm_calls":0}
Acceptance tests — เขียนไว้ ยังไม่ได้รัน
"""Authored checks for the teaching model. Status: NOT RUN."""
import json
import subprocess
import sys
import unittest
from pathlib import Path
from singleflight_mock import ContractError, SingleFlight
class LifecycleTests(unittest.TestCase):
def test_join_fanout_cleanup_and_later_fresh_flight(self):
m = SingleFlight(); self.assertEqual(m.request("A", "W1")["event"], "START")
self.assertEqual(m.request("A", "W2")["event"], "JOIN")
self.assertEqual(m.settle("A", 1, "OK", "catalog price 100 (mock)")["waiters"], ["W1", "W2"])
self.assertEqual(m.request("A", "W3")["flight_id"], 2)
def test_error_fanout_cleanup_without_retry(self):
m = SingleFlight(); m.request("A", "W1")
r = m.settle("A", 1, "ERROR", "mock_unavailable")
self.assertEqual((r["outcome"], r["waiters"]), ("ERROR", ["W1"]))
self.assertEqual(m.snapshot()["active"], []); self.assertEqual(len(m.events), 2)
fresh = m.request("A", "W2")
self.assertEqual((fresh["event"], fresh["flight_id"]), ("START", 2))
def test_stale_id_does_not_remove_new_flight(self):
m = SingleFlight(); m.request("A", "W1"); m.settle("A", 1, "OK", "first"); m.request("A", "W2")
before = m.snapshot()
with self.assertRaises(ContractError): m.settle("A", 1, "OK", "stale")
self.assertEqual(m.snapshot(), before)
def test_scope_partition_and_untrusted_alias(self):
m = SingleFlight(); m.request("A", "W1"); m.request("B", "W2"); before = m.snapshot()
with self.assertRaises(ContractError): m.request("team-a/view-r7", "W3")
self.assertEqual(m.snapshot(), before)
self.assertEqual([x["alias"] for x in before["active"]], ["A", "B"])
def test_duplicate_waiter_and_waiter_capacity(self):
m = SingleFlight()
for w in ("W1", "W2", "W3", "W4"): m.request("A", w)
before = m.snapshot()
for w in ("W1", "W5"):
with self.assertRaises(ContractError): m.request("A", w)
self.assertEqual(m.snapshot(), before)
def test_waiter_cannot_join_two_pending_keys_and_rejection_is_unchanged(self):
m = SingleFlight(); m.request("A", "W1"); before = m.snapshot()
with self.assertRaises(ContractError): m.request("B", "W1")
self.assertEqual(m.snapshot(), before)
def test_returned_delivery_list_does_not_alias_event_history(self):
m = SingleFlight(); m.request("A", "W1"); result = m.settle("A", 1, "OK", "price")
result["waiters"].append("W8")
self.assertEqual(m.events[-1]["waiters"], ["W1"])
self.assertEqual(m.deliveries[-1].waiters, ("W1",))
class CliSchemaTests(unittest.TestCase):
def run_cli(self, value):
p = Path(__file__).with_name("singleflight_mock.py")
raw = value if type(value) is str else json.dumps(value)
return subprocess.run([sys.executable, str(p)], input=raw, text=True, capture_output=True)
def test_missing_extra_and_operation_fields_rejected(self):
cases = ({}, {"events": [], "extra": 1}, {"events": [{"op":"request", "alias":"A"}]},
{"events": [{"op":"request", "alias":"A", "waiter":"W1", "extra":1}]})
for value in cases:
with self.subTest(value=value): self.assertEqual(self.run_cli(value).returncode, 2)
def test_exact_schema_accepted(self):
r = self.run_cli({"events":[{"op":"request", "alias":"A", "waiter":"W1"}]})
self.assertEqual(r.returncode, 0); self.assertEqual(json.loads(r.stdout)["events"][0]["event"], "START")
def test_duplicate_json_fields_rejected(self):
raw = '{"events":[],"events":[{"op":"request","alias":"A","waiter":"W1"}]}'
self.assertEqual(self.run_cli(raw).returncode, 2)
if __name__ == "__main__": unittest.main()
README — Runtime / วิธีใช้ / ขอบเขต
# Single-flight mock
This authored Python 3.12+ standard-library package demonstrates an in-process event/state model for sharing one pending same-key read result. It records intents; it does not perform concurrent execution, call a provider, or authenticate users.
## Runtime
- Python 3.12 or newer; standard library only.
- No dependencies to install and no environment variables.
- No network, API, LLM, filesystem writes, or external service costs.
- Provider Sandbox and Live integration: **NOT IMPLEMENTED**.
## Files
- `singleflight_mock.py`: bounded model and stdin/stdout JSON CLI.
- `test_singleflight_mock.py`: authored lifecycle, failure, key security, capacity, and CLI schema checks.
- `example-input.json`: fixed eight-event scenario.
- `expected-output.json`: hand-authored expected result, not captured from execution.
## Run the example
POSIX:
```sh
python3.12 singleflight_mock.py < example-input.json
```
PowerShell:
```powershell
Get-Content -Raw .\example-input.json | py -3.12 .\singleflight_mock.py
```
Input is exactly `{"events":[...]}` with 1..32 events. A request has exactly `op`, `alias`, and `waiter`; settlement has exactly `op`, `alias`, `flight_id`, `outcome`, and `payload`. The only aliases are fixed trusted registry labels `A` and `B`; waiters are W1..W8 and one label cannot represent two pending requests; outcomes are OK or ERROR; payloads are printable strings 1..80 characters. Duplicate JSON fields are rejected. Invalid input returns JSON error and exit 2.
The scenario has five request events: three START intents, two JOIN intents, and three settlements delivering to five waiters. The model counts zero actual tool/API/LLM calls. A and B share tool/query/schema but have different preapproved scope, so they start separately.
## Authored checks
Testing status: **NOT RUN**. Commands for a reviewer who chooses to run them:
POSIX: `python3.12 -m unittest -v`
PowerShell: `py -3.12 -m unittest -v`
Checks cover lifecycle cleanup, a fresh START after ERROR, stale IDs, scope partition, untrusted alias rejection, global waiter uniqueness, waiter capacity, immutable delivery/event separation, duplicate JSON fields, and CLI missing/extra fields. They are authored checks, not evidence of a passing run. The fixed two-key registry does not exercise the active-flight cap through the CLI.
## Boundaries
The two registry entries represent preauthorized mock read views only. A live adapter must authenticate and authorize every caller before joining, and derive scope from trusted identity/configuration rather than model or tool output. Share results only when every waiter is entitled to equivalent data. Never use this for writes or side effects. Omitting a security or response dimension from the key can expose data.
This is a sequential, in-process model. It has no scheduler, locking, cancellation, queue, retry, timeout, persistence, cross-process coordination, completed-result cache, rate limit, latency measurement, cost-saving or exactly-once guarantee. Capacity values are teaching bounds, not provider quotas.
Failure modes: จุดที่ flight ต้องหยุดหรือแยก
งานที่ไม่ตอบอาจทำให้ flight และผู้รอค้าง. ระบบจริงต้องกำหนด timeout/deadline, settlement และ cancellation แยก; mock ไม่มี clock หรือ scheduler. การยกเลิกผู้รอคนเดียวไม่ควรยกเลิก shared work ที่ยังมีผู้รอคนอื่น.
Process restart หรือหลาย worker ทำให้ in-memory registry หายหรือแยกกัน และอาจเปิด flight ซ้ำ. ระบบข้าม process ต้องออกแบบ atomic shared state, recovery และ stale worker; local ID ไม่ใช่เครื่องมือประสานงาน.
ผล personalize, permission, locale หรือ schema ที่ต่างกันต้องแยก key หรือไม่ coalesce. การ normalize ผิดทำให้ key ชน; waiter เต็มต้องถูกปฏิเสธตาม policy. ห้ามใช้การรวมเพื่อ writes หรืออ้าง exactly-once.
การรวม writes หรือ side effects อาจทำให้ผู้เรียกคนหนึ่งได้รับผลจากการกระทำของอีกคน และไม่ได้ให้ exactly-once. แม้เป็น read, partial delivery หรือ response มาช้าหลัง timeout ต้องมีนโยบายของ adapter และระบบจัดเก็บ; event model นี้ไม่ครอบคลุมผลเหล่านั้น.
Copyable task brief: ทำ read-coalescing mock ที่ตรวจได้
คัดลอก brief หกส่วนด้านล่างเพื่อมอบหมาย Mock นี้. หาก key, scope หรือสิทธิ์ไม่ชัด ให้หยุดถามผู้รับผิดชอบ.
1. เป้าหมาย: สร้างแบบจำลอง Python ที่รวมคำขออ่านซึ่งมี trusted read key เดียวกันเฉพาะระหว่าง flight ยัง pending แล้วส่งผล OK หรือ ERROR ให้ผู้รอและล้าง flight
2. ข้อมูลและเครื่องมือ: Python 3.12+ standard library; event fixture JSON; registry A/B ที่กำหนดใน source; ไม่ใช้ API, LLM, network หรือ dependency ภายนอก
3. ขอบเขตสิทธิ์: อ่านอย่างเดียวและจำลองใน process เดียว; key คือ tool, trusted_scope, normalized_query, schema_version; alias และ waiter มาจากรายการคงที่; ห้ามรับ scope จาก model/tool output; ไม่มี concurrency จริง, retry, queue, cancellation, cache, persistence หรือ deployment
4. สิ่งส่งมอบ: ไฟล์โค้ด, test source, example-input.json, expected-output.json, README และตาราง hand trace ของแปด event พร้อมระบุ call จริงเป็นศูนย์
5. เกณฑ์ตรวจรับ: A/W1 START F1; A/W2,W3 JOIN F1; B/W4 START F2; settle A/F1 OK ส่ง W1-W3 และล้าง A; A/W5 START F3; settle B/F2 ERROR ส่ง W4 ไม่ retry; settle A/F3 OK ส่ง W5; scope แยก, invalid/stale input ไม่ mutate, CLI exact schema
6. จุดหยุด: หยุดหลังไฟล์และ hand trace ตรง contract พร้อม README และ test cases ที่เขียนไว้; ห้าม import, execute, compile, install, เรียก API/LLM, ต่อ Sandbox/Live, ใช้ secret หรือเปลี่ยนระบบจริง; runtime status คง NOT RUNแบบฝึก: เปลี่ยนจังหวะ request แล้วตรวจ key
สร้าง hand trace: A/W1 START, A/W2 JOIN, settle F1 ERROR, A/W3 START F2 แล้ว settle OK. เพิ่ม B/W4, stale ID หรือ waiter ซ้ำ; บันทึก state และ recipients หลังทุก event แล้วอธิบายว่าเหตุใด invalid event ไม่เปลี่ยน state. ใช้ rubric ด้านล่างตรวจ key, cleanup, scope และการแยก Mock/Sandbox/Live.
3D Lab และ Visual Interactive: อ่าน flight และผู้รอ
เลือกเหตุการณ์ในเมนูของ 3D Lab แบบ CSS 3D แล้วดูงานที่ยังค้างใน lane A/B, flight ID และผู้ร่วมรอ ภาพแสดงสถานะหลังเหตุการณ์ที่เลือก: A ว่างหลังเหตุการณ์ 5 และเปิด F3 ในเหตุการณ์ 6 ก่อนล้างเมื่อจบในเหตุการณ์ 8
Visual Interactive แยกให้เลือก Key A/B ของผู้เรียกสามราย และเลือกว่ารับคำขอก่อนงานจบหรือส่งผลก่อนคนถัดไป ดูเจตนา START/JOIN และขอบเขตข้อมูล A/B ค่าที่นับไม่ใช่จำนวน API calls จริง
เมนูและปุ่มใช้แป้นพิมพ์ได้ตามองค์ประกอบ HTML ที่ออกแบบไว้ มี focus, labels, ข้อความผลและตาราง fallback การลด motion อิงค่าที่ตั้งในระบบปฏิบัติการ การทำงานจริงในเบราว์เซอร์ยัง UNKNOWN
3D Lab — งานที่ยังค้างกับผู้ร่วมรอ
แบบจำลองแปดเหตุการณ์: A และ B อ่าน SKU เดียวกัน แต่เป็นขอบเขตข้อมูลคนละทีม จึงไม่รวมเที่ยวกัน กล่อง 3D แสดงรายการที่ยังไม่จบหลังเหตุการณ์ที่เลือก ไม่ใช่คิว Tool จริง และไม่มีการเรียก API
A ยังไม่เคยมีงานค้าง จึงสร้างเจตนาเริ่มงานหนึ่งเที่ยว มี W1 รอผล ไม่ได้เรียก Tool จริง
A: F1 ผู้ร่วมรอ W1 · B: ไม่มีงานค้าง
Key รวม Tool, ขอบเขตที่เชื่อถือได้, คำขอที่กำหนดไว้ และรูปแบบผลลัพธ์ ตัวอย่างนี้ไม่มีระบบตรวจสิทธิ์จริง การเลือก A/B เป็นการเลือกกรณีจำลองที่เตรียมไว้
| เหตุการณ์ | สถานะ A หลังเหตุการณ์ | สถานะ B | ผลส่งให้ผู้ร่วมรอ |
|---|---|---|---|
| 1 START A | F1: W1 | ว่าง | ยังไม่มี |
| 2 JOIN A | F1: W1,W2 | ว่าง | ยังไม่มี |
| 3 JOIN A | F1: W1,W2,W3 | ว่าง | ยังไม่มี |
| 4 START B | F1: W1,W2,W3 | F2: W4 | ยังไม่มี |
| 5 settle A/F1 | ว่าง | F2: W4 | OK → W1,W2,W3 |
| 6 START A ใหม่ | F3: W5 | F2: W4 | ไม่หยิบผลเก่ามาใช้ |
| 7 settle B/F2 | F3: W5 | ว่าง | ERROR → W4 |
| 8 settle A/F3 | ว่าง | ว่าง | OK → W5 |
Visual Interactive — Key กับช่วงเวลาที่ซ้อนกัน
แบบจำลองย่อยสามคำขอ W1,W2,W3: เลือก A หรือ B ซึ่งต่างขอบเขตข้อมูล และเปรียบเทียบช่วงที่งานยังไม่จบกับช่วงที่จบก่อนผู้ขอคนถัดไป ปุ่มเพียงคำนวณเจตนา START/JOIN ไม่เริ่ม Tool และไม่ใช่ผลวัดการประหยัด
| ผู้ขอ | Key | เจตนา | เที่ยว | ผู้ร่วมรอ ณ ตอนรับคำขอ |
|---|---|---|---|---|
| W1 | A | START | F1 | W1 |
| W2 | A | JOIN | F1 | W1,W2 |
| W3 | A | JOIN | F1 | W1,W2,W3 |
โหมดซ้อนกันยังไม่ settle: เมื่อแต่ละเที่ยวจบจึงส่งผลให้ผู้ร่วมรอและลบ entry ส่วนโหมดจบก่อนคนถัดไปจะลบหลังแต่ละคำขอ ทำให้เริ่มใหม่แม้ Key เดิม A/B เป็นขอบเขตสมมติที่เตรียมไว้ ไม่ใช่ระบบตรวจสิทธิ์จริง
เชื่อม Video Lab กับงานเตรียม asset โดยไม่อ้างว่าเป็น single-flight
Video Lab รุ่น active 1.4.0 ที่อ่านผ่าน WordPress ในรอบนี้จำลอง workflow วิดีโอ ไม่ได้ implement single-flight ใช้แผนค้นข้อมูล asset เพื่อพิจารณาว่างานอ่านใดใช้ผลร่วมกันได้เมื่อขอบเขตสิทธิ์และคำขอตรงกัน
อ่าน prompt จาก Lab เทียบกับ brief หกส่วนใกล้เครื่องมือ ระบุว่าใครมีสิทธิ์เห็นผลและกรณีใดต้องแยก Key ค่าเวลาและเงินใน Lab เป็นสมมติฐานที่ปรับได้ ไม่ใช่ผลวัด ไม่มีการเรียก catalog หรือประสานผู้เรียกจริงในบทเรียนนี้
ใช้ Video Lab สำรวจแผนงานและร่าง prompt แล้วตรวจเทียบ brief หกส่วนนี้ กลไก single-flight อยู่ในตัวอย่างแยก ปลั๊กอินไม่ได้รวมคำขออ่าน Tool ให้
1. เป้าหมาย: วางแผนวิดีโอสินค้าและจำลองการรวมคำขออ่านข้อมูลสินค้าเดียวกันระหว่างกำลังเตรียม asset ไม่ผลิตหรือส่งวิดีโอจริง
2. ข้อมูลและเครื่องมือ: Video Lab ใช้สำรวจ workflow และร่าง prompt; ใช้เหตุการณ์ A/W1,A/W2,A/W3 ก่อนผลสมมติพร้อมแล้วจึงส่งผลร่วมกัน ส่วน B เป็นขอบเขตข้อมูลอีกทีมจึงแยกเที่ยว
3. ขอบเขต: ไม่เชื่อมบัญชี ไม่เรียก catalog/โมเดลจริง ไม่สร้าง asset จริง ไม่ติดตั้งโปรแกรมหรือใช้เงินจริง ค่าเวลา/เงินใน Video Lab เป็นสมมติฐานที่ปรับได้ ไม่ใช่ผลวัด
4. สิ่งส่งมอบ: prompt จาก Lab พร้อม brief เสริมเรื่องสิทธิ์ และตาราง START/JOIN/ส่งผล/ลบ entry ที่แยกจากแผนงานวิดีโอ
5. เกณฑ์ตรวจรับ: A ที่มาระหว่างเที่ยวเดิมยังไม่จบใช้ผลร่วมกัน; ต่างขอบเขตไม่รวม; เมื่อจบแล้วคำขอ A ถัดไปเริ่มใหม่ ปลั๊กอินไม่ได้ทำ single-flight ให้
6. จุดหยุด: จบที่แผนและผลจำลอง หยุดถ้าไม่ทราบขอบเขตสิทธิ์หรือความเท่ากันของงานอ่าน ไม่ใช้กับงานเขียนหรือส่งคำสั่งไปบริการจริงจากคลิปดิบ
สู่จักรวาลคอนเทนต์
ChatGPT ตัดต่อวิดีโอแทนคนได้จริงแค่ไหน? สำรวจระบบ AI Video Editing Agent ผ่านฉากสามมิติที่โต้ตอบได้ พร้อมแยกข้อเท็จจริง สร้างพรอมป์ต์ และจำลองต้นทุนก่อนลงมือทำจริง
ออกแบบระบบตัดต่อในมุมมอง 3D
ลากบนฉากเพื่อหมุนมุมมอง แตะโหนดเพื่อดูหน้าที่ของ AI ในแต่ละขั้นตอน
จัดระเบียบไฟล์ ตรวจคุณภาพภาพและเสียง และกำหนดสิทธิ์เข้าถึงให้ AI Agent
แยกความสามารถจริงจากคำโฆษณา
กดแต่ละประเด็นเพื่ออ่านรายละเอียดและแหล่งข้อมูล
ChatGPT Work เข้าถึงไฟล์ในเครื่องได้ตรวจสอบแล้ว / มีเงื่อนไข+
Astra ใช้โปรแกรมคอมพิวเตอร์ทำงานได้ตรวจสอบแล้ว / ไม่ใช่โปรแกรมตัดต่อในตัว+
ตัดเสร็จใน 5–10 นาทีทุกกรณียังพิสูจน์เป็นเกณฑ์ทั่วไปไม่ได้+
ถูกกว่าจ้างคน 10 เท่าต้องกำหนดต้นทุนให้เทียบกันได้+
หนึ่งไอเดีย สามรูปแบบคอนเทนต์
ปรับรูปแบบวิดีโอ แล้วดูกรอบภาพตัวอย่างและ Prompt เปลี่ยนตามแบบเรียลไทม์
ช่วยตัดต่อ?
พร้อมนำไปใช้กับ AI Agent
ตัวสร้าง Prompt นี้ไม่อัปโหลดคลิป ไม่เรียก API เสียเงิน และไม่ตัดต่อวิดีโอจริง ต้องใช้กับเครื่องมือที่ผู้ใช้อนุญาตภายหลัง
จำลองเวลาและต้นทุนของคุณเอง
เลื่อนตัวเลขเพื่อเปรียบเทียบสถานการณ์สมมติ ไม่ใช่ผล benchmark ที่ยืนยันแล้ว
ตัวเลขเป็นตัวอย่างเพื่อช่วยตัดสินใจ ไม่ใช่ผลประหยัดที่รับประกัน
AI ที่ตัดคลิปเป็น ไม่ได้ทำงานลำพัง
สิ่งที่น่าสนใจของ GPT-6 Astra และ GPT-6.1 Sol ไม่ได้อยู่ที่ว่าตัวโมเดลมีปุ่ม Render MP4 ซ่อนอยู่ แต่คือความสามารถในการจัดการงานหลายขั้นตอนร่วมกับเครื่องมืออื่น: อ่านไฟล์ วางลำดับเรื่อง ใช้แอป ตรวจผลลัพธ์ และแก้งานตามข้อเสนอแนะของมนุษย์
สำหรับคนทำคอนเทนต์ ความต่างสำคัญคือการเปลี่ยนจาก “ส่งคลิปให้ AI แล้วหวังว่าจะได้ผลงาน” ไปเป็น Workflow ที่ตรวจรับได้ทุกขั้น ตั้งแต่การเลือกช่วงเด็ด การสร้างซับ การตรวจคำผิด จนถึงการเรนเดอร์ไฟล์ตามสัดส่วนของแต่ละแพลตฟอร์ม
ข้อควรระวังคือระบบ Work ที่รันบนเว็บหรือมือถือไม่ได้หมายความว่า AI เข้าถึงคลิปบนคอมพิวเตอร์บ้านได้ทันที ต้องทำงานในสภาพแวดล้อมที่เปิดให้เข้าถึงไฟล์จริง มีเครื่องมือตัดต่อ และมีสิทธิ์ใช้งานอย่างชัดเจน
รวบรวมไฟล์ในโฟลเดอร์ที่เข้าถึงได้ ระบุสิทธิ์และข้อจำกัด
ให้ Agent เสนอ Hook, ช่วงตัด และโครงเรื่องก่อนตัดต่อ
ตรวจชื่อเฉพาะ ตัวเลข ภาษาไทย และความถูกต้องของคำกล่าวอ้าง
ตรวจความละเอียด เสียง ระยะเวลา safe zone และคำบรรยาย
Security, Cost และการตัดสินใจก่อนใช้ระบบจริง
ตรวจ authentication และ authorization ก่อน JOIN ทุกครั้ง; trusted_scope ต้องมาจาก identity/configuration ไม่ใช่ prompt หรือ Tool output. Key ใช้แบ่งกลุ่ม ไม่ได้อนุญาตสิทธิ์. หากผลเฉพาะบุคคลหรือ policy ต่าง ให้แยก key หรือปิด coalescing.
Mock มีศูนย์ Tool/API/LLM calls จึงไม่มีต้นทุน provider หรือผลประหยัดให้วัด. Sandbox และ Live ไม่ได้ทำ. การลด START ใน fixture ไม่พิสูจน์ cost/latency saving; coordination, storage และ observability จริงอาจมีต้นทุน.
ใช้กับ read-only operation เมื่อทุก caller ได้รับอนุญาตและ response เทียบเท่ากัน. อย่าใช้กับ writes หรืออ้าง exactly-once. ระบบจริงยังต้องกำหนด timeout, partial delivery, persistence, atomic coordination, privacy และ recovery.
จำนวน START ที่น้อยกว่า request intents ในตัวอย่างเป็นเพียงโครงสร้างของ event; ไม่มีการอ่าน network, latency distribution หรือราคาผู้ให้บริการ. อย่ารายงาน cost saving หรือ performance gain จากจำนวน intent. หากจะวัดในระบบจริงให้เก็บจำนวน join, upstream calls, latency และข้อผิดพลาดภายใต้ workload ที่ได้รับอนุญาต.
FAQ
JOIN ต้องรอผลหรือไม่? ใช่; ผู้รอรับผลเมื่อ flight pending settle. Mock ไม่จำลอง async wait.
หลัง settle เก็บ cache หรือไม่? ไม่; entry ถูกลบและ request ใหม่ START flight ใหม่.
ERROR แล้ว retry อัตโนมัติไหม? ไม่; ส่ง error และ cleanup. Caller แยกนโยบาย retry.
ทำไม A/B ไม่รวม? trusted_scope ต่างกัน จึงแยกขอบเขตข้อมูล.
scope เท่ากับ auth หรือไม่? ไม่; ต้อง authorize caller ก่อนทุก join.
มีการทดสอบ code/Lab หรือไม่? Code/tests NOT RUN; Lab UI UNKNOWN.
แหล่งอ้างอิงและบทเรียนต่อเนื่อง
ตรวจต้นทางเมื่อ 22 ตุลาคม 2026 ตัวอย่างโค้ด ตาราง ภาพและ Lab เป็นการออกแบบเพื่อการเรียนรู้ ไม่มีผลทดสอบ API หรือผลวัดประสิทธิภาพ