Exclusive time of functions

A single-threaded program logs "<function>:start:<t>" when a call begins and "<function>:end:<t>" when it ends. Compute each function's exclusive time: time spent in its own body, not in functions it called. Ask:

• Are timestamps points or units? In the classic version, start:3 means the call begins at the start of time unit 3 and end:5 means it ends at the end of unit 5, so it ran for 3 units. Get this convention explicitly; it is the most common off-by-one.
• Identifiers. Integers 0..n-1 with an answer list, or names with an answer dict. Use names; it covers both.
• Recursion. A function can call itself; each call is a separate frame.
• Bad logs. An end that does not match the top frame, or frames still open at the end. Raise, or report? Ask; raising is a fine default.

Calls on one thread nest perfectly, so a stack of open frames models the program.

Keep prev, the moment the timeline was last attributed. Every event closes a segment of time that belongs to whoever is on top of the stack:

• start at t: the current top (if any) ran from prev to t, exclusive of t: add t - prev to it. Push the new frame. Set prev = t.
• end at t: the top ran from prev through the end of unit t: add t - prev + 1. Pop it (checking the name matches). Set prev = t + 1.

Every unit of time is credited to exactly one frame, so the totals add up to the span of the log. That is the invariant to state and to test.

Inclusive time (wall time of each call, children included) comes from the same pass: push t with the frame and, at its end, add t_end - t_start + 1 to an inclusive total, but only for the outermost active call of that function, or recursion double counts.

Part 2: several threads. Logs become "<thread>:<function>:start:<t>", interleaved by time. Frames only nest within a thread, so keep one stack and one prev per thread. With a single CPU you would also need to know which thread was scheduled; with one core per thread, per-thread stacks are enough.

from collections import defaultdict
from typing import Dict, List, Tuple


def _parse(line: str) -> Tuple[str, str, int]:
    name, kind, ts = line.rsplit(":", 2)
    if kind not in ("start", "end"):
        raise ValueError(f"bad event kind in {line!r}")
    return name, kind, int(ts)


def exclusive_times(logs: List[str]) -> Dict[str, int]:
    totals: Dict[str, int] = defaultdict(int)
    stack: List[str] = []
    prev = 0
    for line in logs:
        name, kind, t = _parse(line)
        if kind == "start":
            if stack:
                totals[stack[-1]] += t - prev
            stack.append(name)
            prev = t
        else:
            if not stack or stack[-1] != name:
                raise ValueError(f"end of {name!r} does not match open call {stack[-1:]}")
            totals[stack.pop()] += t - prev + 1
            prev = t + 1
    if stack:
        raise ValueError(f"calls still open at end of log: {stack}")
    return dict(totals)


def inclusive_times(logs: List[str]) -> Dict[str, int]:
    totals: Dict[str, int] = defaultdict(int)
    stack: List[Tuple[str, int]] = []
    depth: Dict[str, int] = defaultdict(int)   # active frames per function
    for line in logs:
        name, kind, t = _parse(line)
        if kind == "start":
            stack.append((name, t))
            depth[name] += 1
        else:
            top, began = stack.pop()
            if top != name:
                raise ValueError(f"end of {name!r} does not match open call {top!r}")
            depth[name] -= 1
            if depth[name] == 0:                # outermost call only
                totals[name] += t - began + 1
    return dict(totals)


def exclusive_times_by_thread(logs: List[str]) -> Dict[str, int]:
    """Part 2: "<thread>:<function>:<start|end>:<t>", one core per thread."""
    totals: Dict[str, int] = defaultdict(int)
    stacks: Dict[str, List[str]] = defaultdict(list)
    prev: Dict[str, int] = defaultdict(int)
    for line in logs:
        thread, rest = line.split(":", 1)
        name, kind, t = _parse(rest)
        stack = stacks[thread]
        if kind == "start":
            if stack:
                totals[stack[-1]] += t - prev[thread]
            stack.append(name)
            prev[thread] = t
        else:
            if not stack or stack[-1] != name:
                raise ValueError(f"thread {thread}: end of {name!r} does not match")
            totals[stack.pop()] += t - prev[thread] + 1
            prev[thread] = t + 1
    if any(stacks.values()):
        raise ValueError("calls still open at end of log")
    return dict(totals)

O(L) time for L log lines and O(d) extra space for the deepest call stack (plus one entry per distinct function in the result). Part 2 is the same with one stack per thread.

1. Real timestamps. With floating-point times in milliseconds, events are instants, not units: drop the + 1 and prev = t + 1 and use t - prev everywhere. Say which convention you are using in the code.
2. Streaming logs. The algorithm is already one pass with O(depth) state, so it runs on an unbounded stream. Emit a function's running total whenever a frame closes.
3. Flame graphs. Key totals by the whole stack path (main;handle;parse) instead of the function name. That is the folded-stack format flame graph tools read.
4. Missing end events (a crash mid-call). Close the open frames at the last timestamp and mark them truncated rather than raising, so a partial log still gives useful numbers.
5. Agent traces. For an AI agent, the same computation splits a conversation's latency into time in the model, in each tool and in the orchestrator itself, which is how you find what to speed up.

Mid-level (L4)

• Uses a stack and gets the classic example right, including the inclusive end timestamp.
• Handles recursion and back-to-back calls.
• Walks through an example by hand before coding.

Senior (L5)

• States the invariant (each time unit credited to exactly one frame) and tests it.
• Validates malformed logs, parses names safely, and computes inclusive time without double counting recursion.
• Clean, single-pass code with O(L) time and O(depth) space.

Staff (L6+)

• Extends to threads with per-thread stacks and explains what single-core scheduling would add.
• Discusses instants versus units, streaming, truncated logs and flame-graph output.
• Connects it to profiling real systems, such as latency breakdowns of agent traces.